website_builder.py

from __future__ import annotations

import html
import mimetypes
import os
import re
import secrets
import shutil
import uuid
from datetime import timedelta
from pathlib import Path, PurePosixPath
from typing import Any

import gradio as gr
from fastapi import HTTPException, Request
from fastapi.responses import FileResponse, HTMLResponse
from starlette.background import BackgroundTask

from core import (
    BACKEND,
    PLANS,
    SETTINGS,
    STORE,
    decrypt_file,
    human_bytes,
    iso,
    parse_dt,
    utcnow,
)

SITE_NAME_RE = re.compile(
    r"^[a-z0-9](?:[a-z0-9-]{0,98}[a-z0-9])?$"
)
TEXT_EXTENSIONS = {
    ".html",
    ".htm",
    ".css",
    ".js",
    ".mjs",
    ".json",
    ".txt",
    ".md",
    ".xml",
    ".svg",
    ".csv",
    ".yml",
    ".yaml",
    ".toml",
    ".ini",
    ".map",
}
MAX_EDITOR_BYTES = 2 * 1024 * 1024


def _new_public_id() -> str:
    used = {
        str(project.get("public_id", ""))
        for project in STORE.state.get("websites", {}).values()
        if isinstance(project, dict)
    }
    value = secrets.token_hex(6)
    while value in used:
        value = secrets.token_hex(6)
    return value


def _new_preview_id() -> str:
    used = {
        str(project.get("preview_id", ""))
        for project in STORE.state.get("websites", {}).values()
        if isinstance(project, dict)
    }
    value = secrets.token_urlsafe(18)
    while value in used:
        value = secrets.token_urlsafe(18)
    return value


def ensure_website_state() -> None:
    changed = False
    with STORE.lock:
        if not isinstance(STORE.state.get("websites"), dict):
            STORE.state["websites"] = {}
            changed = True

        for project_id, project in list(
            STORE.state["websites"].items()
        ):
            if not isinstance(project, dict):
                STORE.state["websites"].pop(project_id, None)
                changed = True
                continue

            project.setdefault("id", project_id)
            project.setdefault("owner_id", "")
            project.setdefault("name", "website")
            project.setdefault("created_at", iso())
            project.setdefault("updated_at", iso())
            project.setdefault("published", False)
            project.setdefault("published_at", None)
            project.setdefault("last_visit_at", None)
            project.setdefault("waking_until", None)
            project.setdefault("closed_at", None)
            project.setdefault("logs", [])

            if not project.get("public_id"):
                project["public_id"] = _new_public_id()
                changed = True

            if not project.get("preview_id"):
                project["preview_id"] = _new_preview_id()
                changed = True

        if changed:
            STORE._save_unlocked(
                "Initialize CloudVault simple website builder"
            )


ensure_website_state()


def _user_and_plan(
    user_id: str | None,
) -> tuple[dict[str, Any], Any]:
    user = STORE.get_user(user_id)
    if not user:
        raise PermissionError("Login required.")

    if user.get("role") == "admin":
        return user, PLANS["admin"]

    plan = PLANS[user.get("plan", "free")]
    if plan.key not in {"plus", "pro"}:
        raise PermissionError(
            "Web Site Builder is available only for Plus and Pro accounts."
        )

    return user, plan


def _site_limit(plan_key: str) -> int:
    if plan_key in {"pro", "admin"}:
        return 3
    return 1


def _sleep_days(plan_key: str) -> int:
    if plan_key == "pro":
        return 7
    if plan_key == "plus":
        return 2
    return 36500


def _append_log_unlocked(
    project: dict[str, Any],
    level: str,
    message: str,
) -> None:
    logs = project.setdefault("logs", [])
    logs.append(
        {
            "time": iso(),
            "level": str(level or "INFO").upper(),
            "message": str(message or ""),
        }
    )
    if len(logs) > 250:
        del logs[:-250]


def validate_site_name(name: str) -> str:
    clean = (name or "").strip().lower()

    if not clean:
        raise ValueError("Enter a website name.")

    if len(clean) > 100:
        raise ValueError(
            "Website names may contain at most 100 characters."
        )

    if not SITE_NAME_RE.fullmatch(clean):
        raise ValueError(
            "Use lowercase English letters, numbers and hyphens only. "
            "The name must start and end with a letter or number."
        )

    return clean


def normalize_web_path(path: str) -> str:
    raw = (path or "").strip().replace("\\", "/").strip("/")

    if not raw:
        raise ValueError("Enter a file name.")

    if len(raw) > 240:
        raise ValueError("The file path is too long.")

    parts = PurePosixPath(raw).parts
    if (
        not parts
        or any(part in {"", ".", ".."} for part in parts)
    ):
        raise ValueError("The file path is invalid.")

    for part in parts:
        if (
            "\x00" in part
            or any(character in part for character in '<>:"|?*')
        ):
            raise ValueError(
                "The file path contains unsupported characters."
            )

    return "/".join(parts)


def _project_files_unlocked(
    project_id: str,
) -> list[dict[str, Any]]:
    return [
        dict(item)
        for item in STORE.state.get("files", {}).values()
        if item.get("website_project_id") == project_id
    ]


def _project_unlocked(
    user_id: str,
    project_id: str,
) -> dict[str, Any]:
    project = STORE.state.get("websites", {}).get(project_id)

    if (
        not project
        or project.get("owner_id") != user_id
    ):
        raise ValueError("Website project not found.")

    return project


def list_user_projects(
    user_id: str | None,
) -> list[dict[str, Any]]:
    user, _plan = _user_and_plan(user_id)
    with STORE.lock:
        projects = [
            dict(project)
            for project in STORE.state.get("websites", {}).values()
            if project.get("owner_id") == user["id"]
        ]
    projects.sort(key=lambda project: project.get("created_at", ""))
    return projects


def get_user_project(
    user_id: str | None,
    project_id: str | None = None,
) -> dict[str, Any] | None:
    projects = list_user_projects(user_id)
    if not projects:
        return None
    if project_id:
        selected = next(
            (project for project in projects if project.get("id") == project_id),
            None,
        )
        if selected:
            return selected
    return projects[0]



def _folder_marker_unlocked(
    project_id: str,
    folder_path: str,
) -> dict[str, Any] | None:
    wanted = folder_path.casefold()
    for item in STORE.state.get("files", {}).values():
        if (
            item.get("website_project_id") == project_id
            and item.get("is_folder")
            and str(item.get("website_path", "")).casefold() == wanted
        ):
            return item
    return None


def create_project_folder(
    user_id: str | None,
    project_id: str | None,
    folder_path: str,
) -> dict[str, Any]:
    user, _plan = _user_and_plan(user_id)
    if not project_id:
        raise ValueError("Create your website first.")

    clean_path = normalize_web_path(folder_path)
    with STORE.lock:
        project = _project_unlocked(user["id"], project_id)
        if _folder_marker_unlocked(project_id, clean_path):
            raise ValueError("This folder already exists.")
        if _find_file_unlocked(project_id, clean_path):
            raise ValueError("A file already uses this name.")

        marker_id = uuid.uuid4().hex
        marker = {
            "id": marker_id,
            "owner_id": user["id"],
            "display_name": PurePosixPath(clean_path).name,
            "remote_path": "",
            "original_size": 0,
            "stored_size": 0,
            "is_folder": True,
            "content_type": "application/x-directory",
            "encryption": "none",
            "created_at": iso(),
            "uploaded_under_plan": user.get("plan", "free"),
            "website_project_id": project_id,
            "website_path": clean_path,
            "website_file": True,
        }
        STORE.state["files"][marker_id] = marker
        project["updated_at"] = iso()
        _append_log_unlocked(project, "INFO", f"Created folder {clean_path}.")
        STORE._save_unlocked("Create CloudVault website folder")
        return dict(marker)


def _directory_entries(
    user_id: str,
    project_id: str,
    directory: str,
) -> list[tuple[str, str]]:
    clean_dir = (directory or "").strip("/")
    prefix = f"{clean_dir}/" if clean_dir else ""
    files = list_project_files(user_id, project_id)

    folders: set[str] = set()
    visible_files: list[str] = []

    for item in files:
        path = str(item.get("website_path", "")).strip("/")
        if not path or not path.startswith(prefix):
            continue

        remainder = path[len(prefix):]
        if not remainder:
            continue

        first, sep, rest = remainder.partition("/")
        if sep:
            folders.add(first)
        elif item.get("is_folder"):
            folders.add(first)
        else:
            visible_files.append(path)

    choices = [
        (f"📁 {name}", f"folder:{prefix}{name}".rstrip("/"))
        for name in sorted(folders, key=str.casefold)
    ]
    choices.extend(
        (f"📄 {PurePosixPath(path).name}", f"file:{path}")
        for path in sorted(visible_files, key=str.casefold)
    )
    return choices


def _parent_directory(directory: str | None) -> str:
    clean = (directory or "").strip("/")
    if not clean:
        return ""
    parent = str(PurePosixPath(clean).parent)
    return "" if parent == "." else parent


def _join_directory(directory: str | None, name: str) -> str:
    clean_dir = (directory or "").strip("/")
    clean_name = normalize_web_path(name)
    return f"{clean_dir}/{clean_name}".strip("/")


def new_file_ui(directory: str | None):
    prefix = (directory or "").strip("/")
    suggested = f"{prefix}/".strip("/") if prefix else ""
    return (
        suggested,
        "",
        "",
        "",
        gr.update(interactive=False),
        "Enter a file name and write the code.",
    )


def create_folder_ui(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
    folder_name: str,
):
    try:
        path = _join_directory(directory, folder_name)
        create_project_folder(user_id, project_id, path)
        return (
            "",
            f"✅ Folder created: {path}",
            *_browser_updates(user_id, project_id, directory),
        )
    except Exception as exc:
        return (
            folder_name,
            f"❌ {exc}",
            *_browser_updates(user_id, project_id, directory),
        )


def _browser_updates(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
):
    try:
        user, _plan = _user_and_plan(user_id)
        clean_dir = (directory or "").strip("/")
        if not project_id:
            return (
                clean_dir,
                gr.update(choices=[], value=None),
                "Root",
            )
        _project_unlocked(user["id"], project_id)
        choices = _directory_entries(user["id"], project_id, clean_dir)
        return (
            clean_dir,
            gr.update(choices=choices, value=None),
            f"/{clean_dir}" if clean_dir else "/",
        )
    except Exception:
        return (
            "",
            gr.update(choices=[], value=None),
            "/",
        )


def open_entry_ui(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
    selected_entry: str | None,
):
    if not selected_entry:
        return (
            *_browser_updates(user_id, project_id, directory),
            "",
            "",
            "",
            "",
            gr.update(interactive=False),
            "Select a file or folder.",
        )

    if selected_entry.startswith("folder:"):
        folder = selected_entry.split(":", 1)[1]
        return (
            *_browser_updates(user_id, project_id, folder),
            "",
            "",
            "",
            "",
            gr.update(interactive=False),
            f"📁 Opened /{folder}",
        )

    file_path = selected_entry.split(":", 1)[1]
    loaded = load_file_ui(user_id, project_id, file_path)
    return (
        *_browser_updates(user_id, project_id, directory),
        *loaded,
    )


def back_folder_ui(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
):
    parent = _parent_directory(directory)
    return _browser_updates(user_id, project_id, parent)


def delete_entry(
    user_id: str | None,
    project_id: str | None,
    selected_entry: str | None,
) -> bool:
    user, _plan = _user_and_plan(user_id)
    if not project_id or not selected_entry:
        raise ValueError("Select a file or folder first.")

    kind, path = selected_entry.split(":", 1)
    clean_path = normalize_web_path(path)
    remote_paths: list[str] = []

    with STORE.lock:
        project = _project_unlocked(user["id"], project_id)
        to_remove: list[str] = []

        for item_id, item in STORE.state.get("files", {}).items():
            if item.get("website_project_id") != project_id:
                continue
            item_path = str(item.get("website_path", "")).strip("/")
            match = (
                item_path.casefold() == clean_path.casefold()
                if kind == "file"
                else (
                    item_path.casefold() == clean_path.casefold()
                    or item_path.casefold().startswith(clean_path.casefold() + "/")
                )
            )
            if match:
                to_remove.append(item_id)
                remote = str(item.get("remote_path", ""))
                if remote:
                    remote_paths.append(remote)

        if not to_remove:
            raise ValueError("The selected item was not found.")

        for item_id in to_remove:
            STORE.state["files"].pop(item_id, None)

        project["updated_at"] = iso()
        _append_log_unlocked(
            project,
            "WARN",
            f"Deleted {'folder' if kind == 'folder' else 'file'} {clean_path}.",
        )
        STORE._save_unlocked("Delete CloudVault website entry")

    if remote_paths:
        BACKEND.delete_many(remote_paths, "Delete CloudVault website entries")
    return True


def delete_entry_ui(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
    selected_entry: str | None,
):
    try:
        delete_entry(user_id, project_id, selected_entry)
        return (
            *_browser_updates(user_id, project_id, directory),
            "",
            "",
            "",
            "",
            gr.update(interactive=False),
            "✅ Deleted.",
        )
    except Exception as exc:
        return (
            *_browser_updates(user_id, project_id, directory),
            "",
            "",
            "",
            "",
            gr.update(interactive=False),
            f"❌ {exc}",
        )


def delete_website_ui(
    user_id: str | None,
    project_id: str | None,
):
    try:
        user, _plan = _user_and_plan(user_id)
        if not project_id:
            raise ValueError("Select a website first.")

        remote_paths: list[str] = []
        with STORE.lock:
            project = _project_unlocked(user["id"], project_id)
            for item_id, item in list(STORE.state.get("files", {}).items()):
                if item.get("website_project_id") == project_id:
                    remote = str(item.get("remote_path", ""))
                    if remote:
                        remote_paths.append(remote)
                    STORE.state["files"].pop(item_id, None)
            STORE.state["websites"].pop(project_id, None)
            STORE._save_unlocked(
                f"Delete CloudVault website {project.get('name', project_id)}"
            )

        if remote_paths:
            BACKEND.delete_many(remote_paths, "Delete CloudVault website project")

        state = builder_state(user_id, None)
        return (*state, "✅ Website and all of its files were deleted.")
    except Exception as exc:
        state = builder_state(user_id, project_id)
        return (*state, f"❌ {exc}")


def upload_to_directory(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
    paths: list[str] | str | None,
    preserve_tree: bool = False,
) -> int:
    user, _plan = _user_and_plan(user_id)
    if not project_id:
        raise ValueError("Create your website first.")
    if not paths:
        return 0

    raw = [paths] if isinstance(paths, str) else [p for p in paths if p]
    readable = [Path(p) for p in raw if Path(p).is_file()]
    if not readable:
        raise ValueError("No readable files were selected.")

    clean_dir = (directory or "").strip("/")
    common = Path(os.path.commonpath([str(p.parent) for p in readable]))
    count = 0
    for path in readable:
        if preserve_tree:
            try:
                relative = path.relative_to(common)
            except ValueError:
                relative = Path(path.name)
            web_path = f"{clean_dir}/{relative.as_posix()}".strip("/")
        else:
            web_path = f"{clean_dir}/{path.name}".strip("/")
        _save_source_file(user["id"], project_id, web_path, path)
        count += 1
    return count


def auto_upload_ui(
    user_id: str | None,
    project_id: str | None,
    directory: str | None,
    paths,
    preserve_tree: bool = False,
):
    try:
        count = upload_to_directory(
            user_id, project_id, directory, paths, preserve_tree
        )
        return (
            None,
            f"✅ {count} item(s) uploaded automatically.",
            *_browser_updates(user_id, project_id, directory),
        )
    except Exception as exc:
        return (
            None,
            f"❌ {exc}",
            *_browser_updates(user_id, project_id, directory),
        )


def _find_file_unlocked(
    project_id: str,
    web_path: str,
) -> dict[str, Any] | None:
    wanted = web_path.casefold()

    for item in STORE.state.get("files", {}).values():
        if (
            item.get("website_project_id") == project_id
            and str(
                item.get("website_path", "")
            ).casefold() == wanted
        ):
            return item

    return None


def list_project_files(
    user_id: str | None,
    project_id: str | None,
) -> list[dict[str, Any]]:
    user, _plan = _user_and_plan(user_id)

    if not project_id:
        return []

    with STORE.lock:
        _project_unlocked(user["id"], project_id)
        files = _project_files_unlocked(project_id)

    return sorted(
        files,
        key=lambda item: str(
            item.get("website_path", "")
        ).casefold(),
    )


def create_project(
    user_id: str | None,
    site_name: str,
) -> dict[str, Any]:
    user, plan = _user_and_plan(user_id)
    clean_name = validate_site_name(site_name)

    with STORE.lock:
        existing = [
            project
            for project in STORE.state.get(
                "websites",
                {},
            ).values()
            if project.get("owner_id") == user["id"]
        ]

        if len(existing) >= _site_limit(plan.key):
            raise ValueError(
                f"Your {plan.name} package allows {_site_limit(plan.key)} "
                f"website{'s' if _site_limit(plan.key) != 1 else ''}."
            )

        project_id = uuid.uuid4().hex
        project = {
            "id": project_id,
            "owner_id": user["id"],
            "name": clean_name,
            "public_id": _new_public_id(),
            "preview_id": _new_preview_id(),
            "created_at": iso(),
            "updated_at": iso(),
            "published": False,
            "published_at": None,
            "last_visit_at": None,
            "waking_until": None,
            "closed_at": None,
            "logs": [
                {
                    "time": iso(),
                    "level": "INFO",
                    "message": (
                        "Website created. The page is blank until code is saved."
                    ),
                }
            ],
        }

        STORE.state["websites"][project_id] = project
        STORE._save_unlocked(
            "Create CloudVault simple website"
        )

        return dict(project)


def _save_source_file(
    user_id: str,
    project_id: str,
    web_path: str,
    source: Path,
    content_type: str | None = None,
) -> dict[str, Any]:
    user, plan = _user_and_plan(user_id)
    clean_path = normalize_web_path(web_path)

    if not source.is_file():
        raise ValueError(
            "The selected file could not be read."
        )

    original_size = source.stat().st_size
    old_remote_path: str | None = None

    with STORE.lock:
        project = _project_unlocked(
            user["id"],
            project_id,
        )
        existing = _find_file_unlocked(
            project_id,
            clean_path,
        )
        old_size = (
            int(existing.get("original_size", 0))
            if existing
            else 0
        )

        user_projected = (
            STORE._used_unlocked(user["id"])
            - old_size
            + original_size
        )

        if (
            user.get("role") != "admin"
            and user_projected > plan.quota_bytes
        ):
            raise ValueError(
                "Your CloudVault storage is full. "
                "Delete files before saving."
            )

        global_projected = (
            STORE._used_unlocked()
            - old_size
            + original_size
        )
        if (
            global_projected
            > SETTINGS.app_storage_limit_bytes
        ):
            raise ValueError(
                "The CloudVault server storage is full."
            )

        if existing:
            old_remote_path = str(
                existing.get("remote_path", "")
            )

    file_id = uuid.uuid4().hex
    remote_path = (
        f"objects/{user['id']}/{file_id}.blob"
    )

    try:
        # Website files are public assets and do not need
        # per-request encryption/decryption.
        BACKEND.upload(
            source,
            remote_path,
            "Save CloudVault website asset",
        )

        with STORE.lock:
            project = _project_unlocked(
                user["id"],
                project_id,
            )
            existing = _find_file_unlocked(
                project_id,
                clean_path,
            )

            if existing:
                STORE.state["files"].pop(
                    existing["id"],
                    None,
                )

            metadata = {
                "id": file_id,
                "owner_id": user["id"],
                "display_name": PurePosixPath(
                    clean_path
                ).name,
                "remote_path": remote_path,
                "original_size": original_size,
                "stored_size": original_size,
                "is_folder": False,
                "content_type": (
                    content_type
                    or mimetypes.guess_type(
                        clean_path
                    )[0]
                    or "application/octet-stream"
                ),
                "encryption": "none",
                "share_password_hash": None,
                "share_token": secrets.token_urlsafe(16),
                "download_count": 0,
                "last_download_at": None,
                "created_at": iso(),
                "uploaded_under_plan": plan.key,
                "website_project_id": project_id,
                "website_path": clean_path,
                "website_file": True,
            }

            STORE.state["files"][file_id] = metadata
            project["updated_at"] = iso()

            _append_log_unlocked(
                project,
                "INFO",
                (
                    f"Saved {clean_path} "
                    f"({human_bytes(original_size)})."
                ),
            )

            if project.get("published"):
                _append_log_unlocked(
                    project,
                    "INFO",
                    "The published website now uses the latest saved files.",
                )

            STORE._save_unlocked(
                "Update CloudVault website asset"
            )

        if (
            old_remote_path
            and old_remote_path != remote_path
        ):
            BACKEND.delete_many(
                [old_remote_path],
                "Replace CloudVault website asset",
            )

        return dict(metadata)

    except Exception:
        BACKEND.delete_many(
            [remote_path],
            "Rollback failed CloudVault website asset",
        )
        raise


def save_text_file(
    user_id: str | None,
    project_id: str | None,
    file_path: str,
    content: str,
) -> dict[str, Any]:
    user, _plan = _user_and_plan(user_id)

    if not project_id:
        raise ValueError("Create your website first.")

    clean_path = normalize_web_path(
        file_path or "index.html"
    )
    temporary = (
        SETTINGS.temp_dir
        / f"website-source-{uuid.uuid4().hex}.txt"
    )

    try:
        temporary.write_text(
            content or "",
            encoding="utf-8",
        )
        return _save_source_file(
            user["id"],
            project_id,
            clean_path,
            temporary,
            (
                mimetypes.guess_type(clean_path)[0]
                or "text/html"
            ),
        )
    finally:
        temporary.unlink(missing_ok=True)


def upload_files(
    user_id: str | None,
    project_id: str | None,
    paths: list[str] | str | None,
) -> int:
    user, _plan = _user_and_plan(user_id)

    if not project_id:
        raise ValueError("Create your website first.")

    if not paths:
        raise ValueError("Choose at least one file.")

    raw_paths = (
        [paths]
        if isinstance(paths, str)
        else [path for path in paths if path]
    )
    readable = [
        Path(path)
        for path in raw_paths
        if Path(path).is_file()
    ]

    if not readable:
        raise ValueError(
            "No readable files were selected."
        )

    count = 0
    for path in readable:
        _save_source_file(
            user["id"],
            project_id,
            path.name,
            path,
        )
        count += 1

    return count


def _read_materialized_file(
    metadata: dict[str, Any],
) -> Path:
    stored = (
        SETTINGS.temp_dir
        / f"website-stored-{uuid.uuid4().hex}.blob"
    )
    suffix = PurePosixPath(
        str(metadata.get("website_path", ""))
    ).suffix
    plain = (
        SETTINGS.temp_dir
        / f"website-plain-{uuid.uuid4().hex}{suffix}"
    )

    if not BACKEND.download(
        metadata["remote_path"],
        stored,
    ):
        raise ValueError(
            "The stored website file could not be downloaded."
        )

    try:
        if metadata.get("encryption", "none") == "none":
            shutil.copy2(stored, plain)
        else:
            decrypt_file(
                stored,
                plain,
                metadata["owner_id"],
                metadata["id"],
                metadata.get("encryption", "none"),
            )
    finally:
        stored.unlink(missing_ok=True)

    return plain


def read_text_file(
    user_id: str | None,
    project_id: str | None,
    web_path: str | None,
) -> str:
    user, _plan = _user_and_plan(user_id)

    if not project_id:
        raise ValueError("Create your website first.")

    clean_path = normalize_web_path(
        web_path or "index.html"
    )

    with STORE.lock:
        _project_unlocked(
            user["id"],
            project_id,
        )
        item = _find_file_unlocked(
            project_id,
            clean_path,
        )

        if not item:
            if clean_path == "index.html":
                return ""
            raise ValueError("File not found.")

        metadata = dict(item)

    if (
        int(metadata.get("original_size", 0))
        > MAX_EDITOR_BYTES
    ):
        raise ValueError(
            "This file is too large for the code editor."
        )

    suffix = PurePosixPath(clean_path).suffix.lower()
    content_type = str(
        metadata.get("content_type", "")
    )

    if (
        suffix not in TEXT_EXTENSIONS
        and not content_type.startswith("text/")
    ):
        raise ValueError(
            "This file type cannot be edited as text."
        )

    plain = _read_materialized_file(metadata)

    try:
        return plain.read_text(
            encoding="utf-8",
            errors="replace",
        )
    finally:
        plain.unlink(missing_ok=True)


def delete_project_file(
    user_id: str | None,
    project_id: str | None,
    web_path: str | None,
) -> bool:
    user, _plan = _user_and_plan(user_id)

    if not project_id or not web_path:
        raise ValueError("Select a file first.")

    clean_path = normalize_web_path(web_path)
    remote_path: str | None = None

    with STORE.lock:
        project = _project_unlocked(
            user["id"],
            project_id,
        )
        item = _find_file_unlocked(
            project_id,
            clean_path,
        )

        if not item:
            raise ValueError("File not found.")

        remote_path = str(
            item.get("remote_path", "")
        )
        STORE.state["files"].pop(
            item["id"],
            None,
        )
        project["updated_at"] = iso()

        _append_log_unlocked(
            project,
            "WARN",
            f"Deleted {clean_path}.",
        )

        STORE._save_unlocked(
            "Delete CloudVault website asset"
        )

    if remote_path:
        BACKEND.delete_many(
            [remote_path],
            "Delete CloudVault website asset",
        )

    return True


def public_site_url(
    project: dict[str, Any],
) -> str:
    base = SETTINGS.public_base_url.rstrip("/")
    return (
        f"{base}/gradio_api/cloudvault-site/"
        f"{project['public_id']}/"
    )


def _project_status(
    project: dict[str, Any],
    plan_key: str,
) -> tuple[str, str]:
    if not project.get("published"):
        return (
            "DRAFT",
            "Saved files are private until Publish Website is pressed.",
        )

    last_visit = parse_dt(
        project.get("last_visit_at")
        or project.get("published_at")
    )

    if last_visit:
        sleeping_after = timedelta(
            days=_sleep_days(plan_key)
        )
        if utcnow() - last_visit >= sleeping_after:
            return (
                "SLEEPING",
                (
                    "The website is sleeping because it has not been visited "
                    f"for {_sleep_days(plan_key)} days."
                ),
            )

    return (
        "PUBLISHED",
        "The website is online and uses the latest saved files.",
    )


def _logs_html(project: dict[str, Any]) -> str:
    logs = list(project.get("logs", []))

    if not logs:
        return (
            '<div class="simple-web-log-empty">'
            "No activity yet."
            "</div>"
        )

    rows: list[str] = []
    for entry in reversed(logs[-100:]):
        created = parse_dt(entry.get("time"))
        shown = (
            created.strftime("%H:%M:%S")
            if created
            else "--:--:--"
        )
        level = html.escape(
            str(entry.get("level", "INFO")).upper()
        )
        message = html.escape(
            str(entry.get("message", ""))
        )

        rows.append(
            f'<div class="simple-web-log-row {level.lower()}">'
            f"<time>{shown}</time>"
            f"<b>{level}</b>"
            f"<span>{message}</span>"
            "</div>"
        )

    return (
        '<div class="simple-web-logs">'
        + "".join(rows)
        + "</div>"
    )


def _status_html(
    status: str,
    message: str,
) -> str:
    class_name = status.lower()
    return (
        f'<div class="simple-web-status {class_name}">'
        "<span></span>"
        "<div>"
        "<small>WEBSITE STATUS</small>"
        f"<b>{html.escape(status)}</b>"
        f"<p>{html.escape(message)}</p>"
        "</div>"
        "</div>"
    )


def _file_choices(
    user_id: str,
    project_id: str,
) -> list[tuple[str, str]]:
    return _directory_entries(user_id, project_id, "")


def builder_state(
    user_id: str | None,
    project_id: str | None = None,
):
    try:
        user, plan = _user_and_plan(user_id)
        projects = list_user_projects(user["id"])
        project = get_user_project(user["id"], project_id)
        project_choices = [
            (str(item.get("name", "Website")), str(item.get("id", "")))
            for item in projects
        ]
        can_create = len(projects) < _site_limit(plan.key)

        if not project:
            return (
                gr.update(visible=True),
                gr.update(visible=False),
                gr.update(choices=[], value=None),
                gr.update(interactive=can_create),
                "",
                "",
                "",
                gr.update(choices=[], value=None),
                "/",
                "",
                "",
                "",
                "",
                "",
                _status_html("DRAFT", "Create your website to begin."),
                '<div class="simple-web-log-empty">No activity yet.</div>',
                gr.update(interactive=False),
                gr.update(interactive=False),
                gr.update(interactive=False),
            )

        with STORE.lock:
            live_project = dict(_project_unlocked(user["id"], project["id"]))

        status, message = _project_status(live_project, plan.key)
        root_entries = _directory_entries(user["id"], live_project["id"], "")
        link = public_site_url(live_project) if live_project.get("published") else ""

        return (
            gr.update(visible=False),
            gr.update(visible=True),
            gr.update(choices=project_choices, value=live_project["id"]),
            gr.update(interactive=can_create),
            live_project["id"],
            live_project["name"],
            "",
            gr.update(choices=root_entries, value=None),
            "/",
            "",
            "",
            "",
            "",
            link,
            _status_html(status, message),
            _logs_html(live_project),
            gr.update(interactive=True),
            gr.update(interactive=bool(live_project.get("published"))),
            gr.update(interactive=True),
        )
    except Exception as exc:
        message = html.escape(str(exc))
        return (
            gr.update(visible=True),
            gr.update(visible=False),
            gr.update(choices=[], value=None),
            gr.update(interactive=False),
            "",
            "",
            "",
            gr.update(choices=[], value=None),
            "/",
            "",
            "",
            "",
            "",
            "",
            _status_html("ERROR", str(exc)),
            '<div class="simple-web-log-row error"><b>ERROR</b><span>'
            + message
            + "</span></div>",
            gr.update(interactive=False),
            gr.update(interactive=False),
            gr.update(interactive=False),
        )


def create_site_ui(
    user_id: str | None,
    site_name: str,
):
    try:
        project = create_project(user_id, site_name)
        state = builder_state(user_id, project["id"])
        return (*state, "", "✅ Website created. The page is blank until code is saved.")
    except Exception as exc:
        state = builder_state(user_id)
        return (*state, site_name, f"❌ {exc}")


def show_create_site_ui():
    return gr.update(visible=True), gr.update(visible=False), ""


def show_code_mode():
    return (
        gr.update(visible=True),
        gr.update(visible=False),
        gr.update(variant="primary"),
        gr.update(variant="secondary"),
    )


def _preview_url(project: dict[str, Any]) -> str:
    base = SETTINGS.public_base_url.rstrip("/")
    return (
        f"{base}/gradio_api/cloudvault-site-preview/"
        f"{project['preview_id']}/"
        f"?v={secrets.token_hex(4)}"
    )


def preview_state(
    user_id: str | None,
    project_id: str | None = None,
):
    try:
        user, _plan = _user_and_plan(user_id)
        project = get_user_project(user["id"], project_id)

        if not project:
            raise ValueError(
                "Create your website first."
            )

        preview_url = _preview_url(project)
        preview_html = (
            '<section class="simple-website-preview">'
            '<div class="simple-preview-bar">'
            "<span></span><span></span><span></span>"
            '<b>1920 × 1080 PREVIEW</b>'
            "</div>"
            f'<iframe src="{html.escape(preview_url, quote=True)}" '
            'width="1920" height="1080"></iframe>'
            "</section>"
        )

        return (
            gr.update(visible=False),
            gr.update(visible=True),
            preview_html,
            "",
            gr.update(variant="secondary"),
            gr.update(variant="primary"),
        )

    except Exception as exc:
        return (
            gr.update(visible=False),
            gr.update(visible=True),
            (
                '<section class="simple-preview-error">'
                "<h3>Preview Error</h3>"
                f"<p>{html.escape(str(exc))}</p>"
                "</section>"
            ),
            f"❌ {exc}",
            gr.update(variant="secondary"),
            gr.update(variant="primary"),
        )


def load_file_ui(
    user_id: str | None,
    project_id: str | None,
    file_path: str | None,
):
    try:
        if not project_id:
            raise ValueError(
                "Create your website first."
            )

        clean_path = normalize_web_path(
            file_path or "index.html"
        )
        content = read_text_file(
            user_id,
            project_id,
            clean_path,
        )

        return (
            clean_path,
            content,
            clean_path,
            content,
            gr.update(interactive=False),
            "✅ File opened.",
        )

    except Exception as exc:
        return (
            file_path or "index.html",
            "",
            "",
            "",
            gr.update(interactive=False),
            f"❌ {exc}",
        )


def editor_changed(
    file_path: str,
    content: str,
    original_path: str,
    original_content: str,
):
    changed = (
        bool((file_path or "").strip())
        and (
            (file_path or "")
            != (original_path or "")
            or (content or "")
            != (original_content or "")
        )
    )

    return gr.update(interactive=changed)


def save_code_ui(
    user_id: str | None,
    project_id: str | None,
    file_path: str,
    content: str,
):
    try:
        saved = save_text_file(
            user_id,
            project_id,
            file_path or "index.html",
            content,
        )
        state = builder_state(user_id, project_id)

        return (
            *state,
            saved["website_path"],
            content,
            saved["website_path"],
            content,
            "✅ Saved. Preview now uses this code.",
        )

    except Exception as exc:
        state = builder_state(user_id, project_id)

        return (
            *state,
            file_path,
            content,
            "",
            "",
            f"❌ {exc}",
        )


def upload_ui(
    user_id: str | None,
    project_id: str | None,
    files,
):
    try:
        count = upload_files(
            user_id,
            project_id,
            files,
        )
        state = builder_state(user_id, project_id)

        return (
            *state,
            None,
            f"✅ {count} file(s) uploaded.",
        )

    except Exception as exc:
        state = builder_state(user_id, project_id)

        return (
            *state,
            files,
            f"❌ {exc}",
        )


def delete_file_ui(
    user_id: str | None,
    project_id: str | None,
    file_path: str | None,
):
    try:
        delete_project_file(
            user_id,
            project_id,
            file_path,
        )
        state = builder_state(user_id, project_id)

        return (
            *state,
            "index.html",
            "",
            "",
            "",
            "✅ File deleted.",
        )

    except Exception as exc:
        state = builder_state(user_id, project_id)

        return (
            *state,
            file_path or "index.html",
            "",
            "",
            "",
            f"❌ {exc}",
        )


def publish_site_ui(
    user_id: str | None,
    project_id: str | None = None,
):
    try:
        user, plan = _user_and_plan(user_id)
        project = get_user_project(user["id"], project_id)

        if not project:
            raise ValueError(
                "Create your website first."
            )

        with STORE.lock:
            live_project = _project_unlocked(
                user["id"],
                project["id"],
            )
            now = iso()
            live_project["published"] = True
            live_project["published_at"] = now
            live_project["last_visit_at"] = now
            live_project["waking_until"] = None
            live_project["closed_at"] = None
            live_project["updated_at"] = now

            _append_log_unlocked(
                live_project,
                "INFO",
                (
                    "Website published. "
                    f"Sleep mode starts after {_sleep_days(plan.key)} "
                    "days without a visit."
                ),
            )

            STORE._save_unlocked(
                "Publish CloudVault website"
            )
            snapshot = dict(live_project)

        status, message = _project_status(
            snapshot,
            plan.key,
        )

        return (
            public_site_url(snapshot),
            _status_html(status, message),
            _logs_html(snapshot),
            "✅ Website published. The link is active now.",
            gr.update(interactive=True),
        )

    except Exception as exc:
        return (
            "",
            _status_html("ERROR", str(exc)),
            (
                '<div class="simple-web-log-row error">'
                f"<b>ERROR</b><span>{html.escape(str(exc))}</span>"
                "</div>"
            ),
            f"❌ {exc}",
            gr.update(interactive=False),
        )


def close_site_ui(
    user_id: str | None,
    project_id: str | None = None,
):
    try:
        user, _plan = _user_and_plan(user_id)
        project = get_user_project(user["id"], project_id)

        if not project:
            raise ValueError(
                "Create your website first."
            )

        with STORE.lock:
            live_project = _project_unlocked(
                user["id"],
                project["id"],
            )
            live_project["published"] = False
            live_project["closed_at"] = iso()
            live_project["waking_until"] = None

            _append_log_unlocked(
                live_project,
                "WARN",
                "Website closed by the owner.",
            )

            STORE._save_unlocked(
                "Close CloudVault website"
            )
            snapshot = dict(live_project)

        return (
            "",
            _status_html(
                "DRAFT",
                "The public website is closed. Saved files remain in storage.",
            ),
            _logs_html(snapshot),
            "✅ Website closed.",
            gr.update(interactive=False),
        )

    except Exception as exc:
        return (
            "",
            _status_html("ERROR", str(exc)),
            (
                '<div class="simple-web-log-row error">'
                f"<b>ERROR</b><span>{html.escape(str(exc))}</span>"
                "</div>"
            ),
            f"❌ {exc}",
            gr.update(interactive=False),
        )


def _blank_html() -> str:
    return (
        "<!doctype html>"
        '<html lang="en">'
        "<head>"
        '<meta charset="utf-8">'
        '<meta name="viewport" '
        'content="width=device-width,initial-scale=1">'
        "<style>"
        "html,body{margin:0;min-height:100%;background:white;}"
        "</style>"
        "</head>"
        "<body></body>"
        "</html>"
    )


def _inject_base(
    document: str,
    base_href: str,
) -> str:
    base_tag = (
        f'<base href="{html.escape(base_href, quote=True)}">'
    )

    if re.search(
        r"<head\b[^>]*>",
        document,
        flags=re.IGNORECASE,
    ):
        return re.sub(
            r"(<head\b[^>]*>)",
            r"\1" + base_tag,
            document,
            count=1,
            flags=re.IGNORECASE,
        )

    return (
        "<!doctype html><html><head>"
        + base_tag
        + "</head><body>"
        + document
        + "</body></html>"
    )


def _lookup_project_by_public_id(
    public_id: str,
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    with STORE.lock:
        project = next(
            (
                dict(candidate)
                for candidate in STORE.state.get(
                    "websites",
                    {},
                ).values()
                if str(
                    candidate.get("public_id", "")
                ) == public_id
            ),
            None,
        )

        if not project:
            return None

        user = STORE.state.get(
            "users",
            {},
        ).get(project.get("owner_id"))

        if not user or user.get("is_banned"):
            return None

        return dict(user), project


def _lookup_project_by_preview_id(
    preview_id: str,
) -> tuple[dict[str, Any], dict[str, Any]] | None:
    with STORE.lock:
        project = next(
            (
                dict(candidate)
                for candidate in STORE.state.get(
                    "websites",
                    {},
                ).values()
                if str(
                    candidate.get("preview_id", "")
                ) == preview_id
            ),
            None,
        )

        if not project:
            return None

        user = STORE.state.get(
            "users",
            {},
        ).get(project.get("owner_id"))

        if not user or user.get("is_banned"):
            return None

        return dict(user), project


def _wake_page(
    public_id: str,
) -> HTMLResponse:
    document = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<meta http-equiv="refresh" content="2;url=/gradio_api/cloudvault-site/{html.escape(public_id, quote=True)}/">
<title>Waking up</title>
<style>
*{{box-sizing:border-box}}
body{{margin:0;min-height:100vh;display:grid;place-items:center;padding:24px;background:#050814;color:white;font-family:Inter,Arial,sans-serif}}
main{{width:min(620px,100%);padding:44px;border:1px solid #2d4167;border-radius:28px;background:linear-gradient(145deg,#111d34,#080e1b);text-align:center;box-shadow:0 35px 100px rgba(0,0,0,.5)}}
i{{display:grid;place-items:center;width:74px;height:74px;margin:0 auto 17px;border-radius:24px;background:linear-gradient(135deg,#7c3aed,#2563eb);font-style:normal;font-size:34px;animation:pulse 1.2s infinite}}
h1{{margin:0 0 8px}}p{{margin:0;color:#9cadc3;line-height:1.7}}
@keyframes pulse{{50%{{transform:scale(1.08)}}}}
</style>
</head>
<body>
<main>
<i>🌐</i>
<h1>Waking up from sleep mode</h1>
<p>The website is starting now. This page will open automatically.</p>
</main>
</body>
</html>"""

    return HTMLResponse(
        document,
        status_code=202,
        headers={"Cache-Control": "no-store"},
    )


def _closed_page() -> HTMLResponse:
    return HTMLResponse(
        """<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Website closed</title>
<style>
body{margin:0;min-height:100vh;display:grid;place-items:center;background:#080b12;color:white;font-family:Arial,sans-serif}
main{text-align:center}p{color:#9aa6b8}
</style>
</head>
<body><main><h1>Website closed</h1><p>The owner has unpublished this website.</p></main></body>
</html>""",
        status_code=404,
        headers={"Cache-Control": "no-store"},
    )


async def _serve_file(
    user: dict[str, Any],
    project: dict[str, Any],
    requested_path: str,
    *,
    preview: bool,
):
    with STORE.lock:
        live_project = _project_unlocked(
            user["id"],
            project["id"],
        )

        if (
            not preview
            and not live_project.get("published")
        ):
            return _closed_page()

        plan_key = (
            "admin"
            if user.get("role") == "admin"
            else user.get("plan", "plus")
        )

        if not preview:
            now = utcnow()
            waking_until = parse_dt(
                live_project.get("waking_until")
            )

            if waking_until and now < waking_until:
                return _wake_page(
                    str(live_project["public_id"])
                )

            last_visit = parse_dt(
                live_project.get("last_visit_at")
                or live_project.get("published_at")
            )

            if (
                last_visit
                and now - last_visit
                >= timedelta(
                    days=_sleep_days(plan_key)
                )
            ):
                live_project["waking_until"] = iso(
                    now + timedelta(seconds=2)
                )
                live_project["last_visit_at"] = iso(now)

                _append_log_unlocked(
                    live_project,
                    "INFO",
                    "Website woke up from sleep mode.",
                )

                STORE._save_unlocked(
                    "Wake CloudVault website"
                )

                return _wake_page(
                    str(live_project["public_id"])
                )

            live_project["waking_until"] = None

            if (
                not last_visit
                or now - last_visit
                >= timedelta(minutes=10)
            ):
                live_project["last_visit_at"] = iso(now)
                STORE._save_unlocked(
                    "Update CloudVault website visit"
                )

        path = (
            requested_path or "index.html"
        ).strip("/")

        if not path:
            path = "index.html"

        try:
            clean_path = normalize_web_path(path)
        except ValueError:
            raise HTTPException(
                status_code=404,
                detail="File not found.",
            )

        item = _find_file_unlocked(
            live_project["id"],
            clean_path,
        )

        if not item and clean_path == "index.html":
            html_items = sorted(
                [
                    candidate
                    for candidate in _project_files_unlocked(live_project["id"])
                    if not candidate.get("is_folder")
                    and PurePosixPath(
                        str(candidate.get("website_path", ""))
                    ).suffix.lower() in {".html", ".htm"}
                ],
                key=lambda candidate: str(
                    candidate.get("website_path", "")
                ).casefold(),
            )
            if html_items:
                item = html_items[0]
                clean_path = str(item.get("website_path", ""))
            else:
                return HTMLResponse(
                    _blank_html(),
                    headers={"Cache-Control": "no-store"},
                )

        if not item:
            raise HTTPException(
                status_code=404,
                detail="File not found.",
            )

        metadata = dict(item)
        public_id = str(
            live_project.get("public_id", "")
        )
        preview_id = str(
            live_project.get("preview_id", "")
        )

    plain = _read_materialized_file(metadata)
    media_type = (
        metadata.get("content_type")
        or mimetypes.guess_type(clean_path)[0]
        or "application/octet-stream"
    )

    if (
        PurePosixPath(clean_path).suffix.lower()
        in {".html", ".htm"}
    ):
        try:
            document = plain.read_text(
                encoding="utf-8",
                errors="replace",
            )
        finally:
            plain.unlink(missing_ok=True)

        base_href = (
            f"/gradio_api/cloudvault-site-preview/"
            f"{preview_id}/"
            if preview
            else (
                f"/gradio_api/cloudvault-site/"
                f"{public_id}/"
            )
        )

        if not document.strip():
            document = _blank_html()
        else:
            document = _inject_base(
                document,
                base_href,
            )

        return HTMLResponse(
            document,
            headers={"Cache-Control": "no-store"},
        )

    def cleanup() -> None:
        plain.unlink(missing_ok=True)

    return FileResponse(
        plain,
        media_type=media_type,
        headers={
            "Cache-Control": "no-store, max-age=0",
            "X-Content-Type-Options": "nosniff",
        },
        background=BackgroundTask(cleanup),
    )


async def public_site_root(
    public_id: str,
):
    lookup = _lookup_project_by_public_id(
        public_id
    )
    if not lookup:
        raise HTTPException(
            status_code=404,
            detail="Website not found.",
        )

    user, project = lookup
    return await _serve_file(
        user,
        project,
        "index.html",
        preview=False,
    )


async def public_site_asset(
    public_id: str,
    path: str,
):
    lookup = _lookup_project_by_public_id(
        public_id
    )
    if not lookup:
        raise HTTPException(
            status_code=404,
            detail="Website not found.",
        )

    user, project = lookup
    return await _serve_file(
        user,
        project,
        path,
        preview=False,
    )


async def preview_site_root(
    preview_id: str,
):
    lookup = _lookup_project_by_preview_id(
        preview_id
    )
    if not lookup:
        raise HTTPException(
            status_code=404,
            detail="Preview not found.",
        )

    user, project = lookup
    return await _serve_file(
        user,
        project,
        "index.html",
        preview=True,
    )


async def preview_site_asset(
    preview_id: str,
    path: str,
):
    lookup = _lookup_project_by_preview_id(
        preview_id
    )
    if not lookup:
        raise HTTPException(
            status_code=404,
            detail="Preview not found.",
        )

    user, project = lookup
    return await _serve_file(
        user,
        project,
        path,
        preview=True,
    )


def register_website_routes(demo) -> None:
    existing = {
        getattr(route, "path", "")
        for route in demo.app.routes
    }

    routes = [
        (
            "/gradio_api/cloudvault-site/{public_id}/",
            public_site_root,
        ),
        (
            "/gradio_api/cloudvault-site/{public_id}/{path:path}",
            public_site_asset,
        ),
        (
            "/gradio_api/cloudvault-site-preview/{preview_id}/",
            preview_site_root,
        ),
        (
            "/gradio_api/cloudvault-site-preview/{preview_id}/{path:path}",
            preview_site_asset,
        ),
    ]

    for path, endpoint in routes:
        if path not in existing:
            demo.app.add_api_route(
                path,
                endpoint,
                methods=["GET", "HEAD"],
                include_in_schema=False,
            )