#!/usr/bin/env python3
"""主题生成器的本地预览服务 + 素材上传代理。

为什么必须起服务：生成器不再自己画一个"仿制页面"，而是把项目真实渲染出来的
页面（outputs/preview-*.html）放进 iframe 当画板，再往里注入正在编辑的主题 CSS。
file:// 下浏览器不允许父页面访问 iframe 的 DOM（即使同目录），预览就只能退回示意图。
起一个本地 HTTP 服务后父子同源，拖拽、注入、实时改令牌都能正常工作。

为什么上传也走这个服务：浏览器直连对象存储要配 CORS、还要把密钥交给页面里的
fetch，各家 S3 兼容实现的 CORS 配置界面又都不一样。放到本机 Python 里代发，
生成器只跟 127.0.0.1 说话：不碰 CORS，密钥不落盘、不进工程文件、不进导出 zip。

用法（在项目根目录执行）：

    python dev/theme-builder/serve.py                   # 默认 0.0.0.0:8777，同局域网可访问
    python dev/theme-builder/serve.py --host 127.0.0.1  # 只给本机（公共网络用这个）
    python -m http.server 8777                          # 只有静态预览，没有上传能力

启动时终端会列出本机与各网卡的地址，照着打开即可。

这个脚本只读文件，不写任何东西，也不改项目配置。
"""

import argparse
import base64
import datetime
import functools
import hashlib
import hmac
import http.server
import json
import os
import re
import socket
import socketserver
import sys
import urllib.error
import urllib.parse
import urllib.request

# dev/theme-builder/serve.py → 上两级就是项目根
PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# 单次请求体上限：一次上传十几张 1600px 的 WebP，base64 后约 1.4 倍体积，
# 64MB 足够，同时挡住误连到公网时的内存放大。
MAX_BODY_BYTES = 64 * 1024 * 1024

EMPTY_SHA256 = hashlib.sha256(b"").hexdigest()


# ---------------------------------------------------------------------------
# SigV4 签名
# ---------------------------------------------------------------------------
# 不用 boto3：这个脚本的定位是「零依赖、拷走就能跑」，装一个 SDK 只为签四个请求
# 不划算。下面是 AWS 文档里 SigV4 for S3 的最小实现，覆盖 PUT / GET / DELETE / HEAD。

def _hmac(key: bytes, msg: str) -> bytes:
    return hmac.new(key, msg.encode("utf-8"), hashlib.sha256).digest()


def _sha256(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def _quote_key(key: str) -> str:
    """按 AWS 规则编码对象键：保留 `/`，其余非 unreserved 字符全部百分号编码。

    用 quote(safe='/-_.~') 而不是逐段 quote：两者等价，但后者容易漏掉空段
    （`a//b`）导致的路径塌缩。S3 的签名 URI 必须与请求行完全一致，差一个字符
    就是 SignatureDoesNotMatch。
    """
    return urllib.parse.quote(key, safe="/-_.~")


def _quote_query(s):
    """query 的编码规则比 path 更严：`/` 也要编码成 %2F，只有 A-Za-z0-9-_.~ 原样保留。"""
    return urllib.parse.quote(str(s), safe="-_.~")


def canonical_query(params):
    """按 AWS 规则拼 canonical query：先按键排序，同名多值再按值排序，逐项编码。

    ListObjectsV2 的 `list-type` / `prefix` / `continuation-token` 全走这里。
    顺序错、或漏了编码都是 SignatureDoesNotMatch —— 而 S3 不会告诉你差在哪。
    """
    pairs = []
    for k, v in (params or {}).items():
        for one in (v if isinstance(v, (list, tuple)) else [v]):
            pairs.append((_quote_query(k), _quote_query(one)))
    pairs.sort()
    return "&".join("%s=%s" % kv for kv in pairs)


def sign_v4(cfg, method, canonical_path, payload_hash, amz_date, host, query=""):
    region = cfg.region
    date = amz_date[:8]
    signed_headers = "host;x-amz-content-sha256;x-amz-date"
    canonical_headers = (
        "host:" + host + "\n"
        "x-amz-content-sha256:" + payload_hash + "\n"
        "x-amz-date:" + amz_date + "\n"
    )
    # CanonicalHeaders 自身每行以 \n 结尾，与 SignedHeaders 之间因此有一个空行，
    # 这是规范要求的形态，不是笔误。
    canonical_request = "\n".join([
        method, canonical_path, query or "", canonical_headers, signed_headers, payload_hash,
    ])
    scope = date + "/" + region + "/s3/aws4_request"
    string_to_sign = "\n".join([
        "AWS4-HMAC-SHA256", amz_date, scope, _sha256(canonical_request.encode("utf-8")),
    ])

    k = _hmac(("AWS4" + cfg.secret_key).encode("utf-8"), date)
    k = _hmac(k, region)
    k = _hmac(k, "s3")
    k = _hmac(k, "aws4_request")
    signature = hmac.new(k, string_to_sign.encode("utf-8"), hashlib.sha256).hexdigest()

    return (
        "AWS4-HMAC-SHA256 Credential=" + cfg.access_key + "/" + scope
        + ", SignedHeaders=" + signed_headers + ", Signature=" + signature
    )


class S3Error(Exception):
    def __init__(self, step, status, code, message):
        Exception.__init__(self, message or code or ("HTTP %s" % status))
        self.step = step
        self.status = status
        self.code = code
        self.message = message or ""


# 常见错误 → 人话。签名失败是最常见的，原因几乎总是密钥/region/时钟三者之一。
S3_HINTS = {
    "SignatureDoesNotMatch": "签名不匹配：核对 Secret Key 是否复制完整、Region 是否与桶一致（R2 填 auto），以及本机时间偏差是否超过 15 分钟。",
    "InvalidAccessKeyId": "Access Key 无效：核对 ID，或该密钥没有这个桶的权限。",
    "AccessDenied": "权限不足：这个密钥没有写权限，或桶策略禁止了该操作。",
    "NoSuchBucket": "桶不存在：核对桶名与 Endpoint 是否属于同一账号/区域。",
    "PermanentRedirect": "Endpoint 与桶所在区域不匹配：换对该区域的 Endpoint（或改用虚拟主机风格）。",
    "AuthorizationHeaderMalformed": "Region 写错了：填桶实际所在区域（R2 用 auto，MinIO 常为 us-east-1）。",
    "InvalidBucketName": "桶名不合法：S3 桶名只允许小写字母、数字、点和中划线。",
    "RequestTimeTooSkewed": "本机时间与服务器偏差过大：校准系统时间后重试。",
}


def _parse_s3_error(body):
    text = body[:2000].decode("utf-8", "replace")
    code = re.search(r"<Code>([^<]+)</Code>", text)
    msg = re.search(r"<Message>([^<]+)</Message>", text)
    return (code.group(1).strip() if code else ""), (msg.group(1).strip() if msg else text.strip()[:200])


class S3Config:
    """一次上传用到的全部参数。只活在内存里，不落盘。"""

    def __init__(self, d):
        ep = str(d.get("endpoint") or "").strip()
        if not ep:
            raise ValueError("Endpoint 没填")
        if "://" not in ep:
            ep = "https://" + ep
        u = urllib.parse.urlsplit(ep)
        if u.scheme not in ("http", "https"):
            raise ValueError("Endpoint 只支持 http / https")
        if not u.hostname:
            raise ValueError("Endpoint 里没有主机名")
        # 路径部分一律忽略：Endpoint 只表达「服务在哪」，桶和键由字段决定。
        self.scheme = u.scheme
        self.port = u.port
        self.host_only = u.hostname
        self.region = str(d.get("region") or "").strip() or "us-east-1"
        self.bucket = str(d.get("bucket") or "").strip()
        if not self.bucket:
            raise ValueError("桶名没填")
        self.prefix = str(d.get("prefix") or "").strip().strip("/")
        self.access_key = str(d.get("accessKey") or "").strip()
        self.secret_key = str(d.get("secretKey") or "").strip()
        if not self.access_key or not self.secret_key:
            raise ValueError("Access Key / Secret Key 没填")
        self.path_style = bool(d.get("pathStyle", True))

    @property
    def host(self):
        if self.path_style:
            return self.host_only if not self.port else "%s:%d" % (self.host_only, self.port)
        return "%s.%s" % (self.bucket, self.host_only) + (":%d" % self.port if self.port else "")

    def object_path(self, key):
        """S3 的签名 URI 必须与请求行逐字一致，所以两者共用这一个函数。"""
        path = ("/" + self.bucket + "/" + key) if self.path_style else ("/" + key)
        return _quote_key(path)

    def url(self, path):
        return "%s://%s%s" % (self.scheme, self.host, path)


def _now():
    return datetime.datetime.now(datetime.timezone.utc)


def s3_request(cfg, method, key="", body=b"", content_type=None, cache_control=None,
               query=None, timeout=25):
    """发一个签名请求。key 为空表示桶级操作（HEAD bucket / ListObjectsV2）。

    返回 (status, headers, body)；4xx/5xx 不抛异常，交由调用方判断 —— 测试连接
    需要把「403 是权限问题」和「连不上是网络问题」区分开讲。
    """
    if key:
        path = cfg.object_path(key)
    else:
        path = ("/" + cfg.bucket) if cfg.path_style else "/"
    qs = canonical_query(query)
    payload_hash = _sha256(body) if body else EMPTY_SHA256
    amz_date = _now().strftime("%Y%m%dT%H%M%SZ")
    auth = sign_v4(cfg, method, path, payload_hash, amz_date, cfg.host, qs)

    url = cfg.url(path) + (("?" + qs) if qs else "")
    req = urllib.request.Request(url, data=body if body else None, method=method)
    req.add_header("x-amz-content-sha256", payload_hash)
    req.add_header("x-amz-date", amz_date)
    req.add_header("Authorization", auth)
    if content_type:
        req.add_header("Content-Type", content_type)
    if cache_control:
        req.add_header("Cache-Control", cache_control)

    try:
        with urllib.request.urlopen(req, timeout=timeout) as r:
            return r.status, dict(r.headers), r.read()
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read()
    except Exception as e:  # URLError / timeout / SSL 错误
        raise S3Error("connect", 0, "NetworkError", "%s（%s）" % (e, cfg.host))


def full_key(cfg, key):
    key = str(key or "").strip().lstrip("/")
    if not key:
        raise ValueError("对象键为空")
    if ".." in key.split("/") or "\\" in key:
        raise ValueError("对象键不能包含 .. 或反斜杠：%s" % key)
    return (cfg.prefix + "/" + key) if cfg.prefix else key


def probe_key(cfg):
    stamp = _now().strftime("%Y%m%dT%H%M%S")
    return full_key(cfg, ".tb-probe-%s.txt" % stamp)


def _public_read(url, timeout=10):
    """匿名 GET 一次，判断「传上去的图能不能直接被浏览器读到」。

    这一步不是必须的（很多人会用自定义域/CDN），但它能把最常见的坑当场暴露：
    桶是私有的，上传全成功、页面里全是 403。
    """
    try:
        with urllib.request.urlopen(url, timeout=timeout) as r:
            return r.status == 200
    except urllib.error.HTTPError as e:
        return e.code == 200
    except Exception:
        return False


def test_connection(cfg):
    steps = []

    status, _, body = s3_request(cfg, "HEAD")
    if status >= 400:
        code, msg = _parse_s3_error(body)
        if not code:
            # HEAD 响应按规范没有 body，S3 的错误码拿不到。只能按状态码归因，
            # 否则前端只能显示一句「HTTP 403」，用户完全不知道该改哪。
            code = {403: "AccessDenied", 404: "NoSuchBucket",
                    301: "PermanentRedirect", 307: "PermanentRedirect",
                    400: "BadRequest"}.get(status, "HTTP%d" % status)
            msg = "桶级探测失败（HEAD 无响应体，按状态码判断）"
        raise S3Error("HeadBucket", status, code, msg)
    steps.append({"name": "连接与凭据", "ok": True, "detail": "桶可访问（HEAD %s）" % cfg.bucket})

    key = probe_key(cfg)
    payload = b"TamaBox theme builder probe\n"
    status, _, body = s3_request(cfg, "PUT", key, payload, "text/plain; charset=utf-8", "no-store")
    if status >= 400:
        code, msg = _parse_s3_error(body)
        raise S3Error("PutObject", status, code, msg)
    steps.append({"name": "写入权限", "ok": True, "detail": "已写入探针 %s" % key})

    status, _, body = s3_request(cfg, "GET", key)
    ok = status == 200 and body == payload
    steps.append({
        "name": "读回校验", "ok": ok,
        "detail": "读回一致" if ok else "读回失败（HTTP %s）" % status,
    })

    public = _public_read(cfg.url(cfg.object_path(key)))
    steps.append({
        "name": "匿名可读", "ok": True,
        "detail": "该路径可直接匿名访问" if public
        else "匿名访问被拒 —— 桶没开公开读时正常，靠自定义域或 CDN 分发即可",
        "warn": not public,
    })

    s3_request(cfg, "DELETE", key)
    return {"ok": True, "steps": steps, "key": key}


# ---------------------------------------------------------------------------
# 上传
# ---------------------------------------------------------------------------

def upload_files(cfg, files):
    if not files:
        raise ValueError("没有要上传的素材")
    if len(files) > 200:
        raise ValueError("一次最多 200 个文件")

    results = []
    for f in files:
        name = str(f.get("name") or f.get("key") or "?")
        try:
            key = full_key(cfg, f.get("key") or f.get("name"))
            data = base64.b64decode(f.get("data") or "", validate=False)
            if not data:
                raise ValueError("内容为空")
            ctype = str(f.get("contentType") or "application/octet-stream")
            # 素材键里带内容版本（导出时由 id 决定），内容不会原地变化 → immutable。
            # 但清单/说明这类「键不变、内容会变」的文件不能 immutable，否则一年内看不到更新，
            # 所以允许逐个文件覆盖。
            cache = str(f.get("cacheControl") or "public, max-age=31536000, immutable")
            status, _, body = s3_request(cfg, "PUT", key, data, ctype, cache)
            if status >= 400:
                code, msg = _parse_s3_error(body)
                results.append({"key": key, "name": name, "ok": False,
                                "status": status, "code": code, "error": msg})
            else:
                results.append({"key": key, "name": name, "ok": True, "bytes": len(data)})
        except Exception as e:
            results.append({"key": name, "name": name, "ok": False, "error": str(e)})

    failed = [r for r in results if not r["ok"]]
    return {"ok": not failed, "files": results, "failed": len(failed), "total": len(results)}


# ---------------------------------------------------------------------------
# 远端清单 / 删除
# ---------------------------------------------------------------------------
def _local(tag):
    """去掉 XML 命名空间。各家 S3 兼容实现都带 `http://s3.amazonaws.com/doc/2006-03-01/`，
    个别（老 MinIO / 某些网关）又不带 —— 只比 localname 两边都能认。"""
    return tag.rsplit("}", 1)[-1]


def parse_list_xml(body):
    """解析 ListObjectsV2 响应。返回 (对象数组, 下一页 token 或 None)。"""
    import xml.etree.ElementTree as ET

    root = ET.fromstring(body)
    objects, token, truncated = [], None, False
    for node in root:
        name = _local(node.tag)
        if name == "Contents":
            row = {}
            for f in node:
                row[_local(f.tag)] = f.text
            objects.append({
                "key": row.get("Key") or "",
                "size": int(row.get("Size") or 0),
                "modified": row.get("LastModified") or "",
                "etag": (row.get("ETag") or "").strip('"'),
            })
        elif name == "NextContinuationToken":
            token = node.text
        elif name == "IsTruncated":
            truncated = (node.text or "").strip().lower() == "true"
    return objects, (token if truncated else None)


def list_objects(cfg, prefix="", max_pages=20, max_keys=1000):
    """列 `<cfg.prefix>/<prefix>` 下所有对象。分页上限只是防呆，正常一个主题不到 100 个。"""
    key_prefix = full_key(cfg, prefix) if str(prefix or "").strip() else (cfg.prefix + "/" if cfg.prefix else "")

    objects, token, pages = [], None, 0
    while pages < max_pages:
        pages += 1
        query = {"list-type": "2", "max-keys": str(max_keys)}
        if key_prefix:
            query["prefix"] = key_prefix
        if token:
            query["continuation-token"] = token
        status, _, body = s3_request(cfg, "GET", query=query)
        if status >= 400:
            code, msg = _parse_s3_error(body)
            raise S3Error("ListObjects", status, code or "HTTP%d" % status, msg)
        page, token = parse_list_xml(body)
        objects.extend(page)
        if not token:
            break
    return {"ok": True, "prefix": key_prefix, "objects": objects, "truncated": bool(token)}


def delete_objects(cfg, keys):
    """逐个 DeleteObject。

    没用批量 POST ?delete：那要额外算 Content-MD5、拼 XML body，而一次最多几十个键，
    省下的往返抵不上多一条签名路径的风险。S3 的 DELETE 本身幂等（不存在的键也回 204）。
    """
    if not keys:
        raise ValueError("没有要删除的对象")
    if len(keys) > 500:
        raise ValueError("一次最多删 500 个对象")

    results = []
    for k in keys:
        try:
            key = full_key(cfg, k)
            status, _, body = s3_request(cfg, "DELETE", key)
            if status >= 400 and status != 404:
                code, msg = _parse_s3_error(body)
                results.append({"key": key, "ok": False, "status": status,
                                "code": code or "HTTP%d" % status, "error": msg})
            else:
                results.append({"key": key, "ok": True})
        except Exception as e:
            results.append({"key": str(k), "ok": False, "error": str(e)})

    failed = [r for r in results if not r["ok"]]
    return {"ok": not failed, "files": results, "failed": len(failed), "total": len(results)}


# ---------------------------------------------------------------------------
# HTTP 层
# ---------------------------------------------------------------------------

class Handler(http.server.SimpleHTTPRequestHandler):
    server_version = "TamaBoxThemeBuilder/1"

    def end_headers(self):
        # 预览页是刚由 go test 生成的，禁止缓存，否则改了 CSS 看不到变化
        self.send_header("Cache-Control", "no-store, must-revalidate")
        super().end_headers()

    def log_message(self, fmt, *args):
        # 只报 4xx/5xx，正常请求不刷屏
        status = args[1] if len(args) > 1 else ""
        if str(status).startswith(("4", "5")):
            sys.stderr.write("  %s\n" % (fmt % args))

    def _json(self, payload, status=200):
        data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
        self.send_response(status)
        self.send_header("Content-Type", "application/json; charset=utf-8")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)

    def _read_json(self):
        n = int(self.headers.get("Content-Length") or 0)
        if n <= 0:
            raise ValueError("请求体为空")
        if n > MAX_BODY_BYTES:
            raise ValueError("请求体过大（%d 字节），一次别传太多素材" % n)
        return json.loads(self.rfile.read(n).decode("utf-8"))

    def _same_origin(self):
        """写接口只接受「页面自己发来的」请求。

        监听 0.0.0.0 之后，同网段里的另一个网页可以借访客浏览器向这里发写请求
        （删对象、改前缀）。浏览器跨站请求一定带 Origin，比对它与 Host 是否同源
        就能挡掉这类「借刀」调用。curl / 脚本不带 Origin，照常放行。
        """
        origin = self.headers.get("Origin")
        if not origin:
            return True
        host = (self.headers.get("Host") or "").strip().lower()
        return urllib.parse.urlsplit(origin).netloc.lower() == host

    def do_GET(self):
        if self.path.split("?")[0] == "/api/s3/ping":
            # 生成器靠它判断「当前是不是 serve.py 起的」——python -m http.server 没有这个端点。
            return self._json({"ok": True, "upload": True})
        return super().do_GET()

    def do_POST(self):
        route = self.path.split("?")[0]
        if route not in ("/api/s3/test", "/api/s3/upload", "/api/s3/list", "/api/s3/delete"):
            return self._json({"ok": False, "error": "未知接口 %s" % route}, 404)
        if not self._same_origin():
            return self._json({"ok": False, "error": "拒绝跨站写请求（Origin 与访问地址不一致）"}, 403)
        try:
            payload = self._read_json()
        except Exception as e:
            return self._json({"ok": False, "error": "请求解析失败：%s" % e})

        try:
            cfg = S3Config(payload.get("config") or {})
        except ValueError as e:
            return self._json({"ok": False, "error": str(e)})

        try:
            if route == "/api/s3/test":
                return self._json(test_connection(cfg))
            if route == "/api/s3/list":
                return self._json(list_objects(cfg, payload.get("prefix") or ""))
            if route == "/api/s3/delete":
                return self._json(delete_objects(cfg, payload.get("keys") or []))
            return self._json(upload_files(cfg, payload.get("files") or []))
        except S3Error as e:
            return self._json({
                "ok": False, "step": e.step, "status": e.status, "code": e.code,
                "error": e.message, "hint": S3_HINTS.get(e.code, ""),
            })
        except ValueError as e:
            return self._json({"ok": False, "error": str(e)})
        except Exception as e:  # 兜底：别把本地服务弄崩，把异常原样告诉页面
            return self._json({"ok": False, "error": "%s: %s" % (type(e).__name__, e)})


def local_addresses():
    """本机所有可用于访问的 IPv4（回环排最后）。

    绑 0.0.0.0 时监听地址本身不能直接点开（浏览器访问 http://0.0.0.0 行为各家不一），
    所以启动时把真实 IP 列出来，省得人去 ipconfig 里翻。
    """
    out = []
    try:
        for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET):
            ip = info[4][0]
            if ip not in out and not ip.startswith("127."):
                out.append(ip)
    except Exception:
        pass
    try:
        # 不联网也会返回默认路由网卡的地址；这一步能补上 getaddrinfo 漏掉的情况
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect(("8.8.8.8", 80))
        ip = s.getsockname()[0]
        s.close()
        if ip and not ip.startswith("127.") and ip not in out:
            out.append(ip)
    except Exception:
        pass
    return out


def main():
    ap = argparse.ArgumentParser(description="TamaBox 主题生成器的本地预览服务")
    ap.add_argument("--port", type=int, default=8777, help="端口，默认 8777")
    ap.add_argument("--host", default="0.0.0.0",
                    help="监听地址，默认 0.0.0.0（同一局域网都能打开）；只给本机用就写 127.0.0.1")
    args = ap.parse_args()

    if not os.path.isdir(os.path.join(PROJECT_ROOT, "outputs")):
        sys.stderr.write(
            "还没生成预览快照。先在项目根执行：\n"
            "    TEST_PREVIEW=1 go test ./templates -run TestGenPreview\n"
        )

    wildcard = args.host in ("0.0.0.0", "::")
    if not wildcard and args.host not in ("127.0.0.1", "localhost", "::1"):
        sys.stderr.write("⚠️ 只监听 %s，仅该地址可达。\n" % args.host)

    handler = functools.partial(Handler, directory=PROJECT_ROOT)
    socketserver.ThreadingTCPServer.allow_reuse_address = True
    # Threading：上传一张几 MB 的图会占住连接，单线程时 iframe 预览和其它请求都得排队。
    with socketserver.ThreadingTCPServer((args.host, args.port), handler) as httpd:
        httpd.daemon_threads = True
        print("项目根：%s" % PROJECT_ROOT)
        if wildcard:
            for ip in local_addresses():
                print("生成器：http://%s:%d/dev/theme-builder/" % (ip, args.port))
            print("生成器：http://127.0.0.1:%d/dev/theme-builder/  （本机）" % args.port)
            print("提示：局域网里的其它设备用上面的 IP 地址打开；手机与电脑要在同一网段。")
            sys.stderr.write(
                "⚠️ 正在对局域网开放（%s:%d）。上传接口没有账号体系 —— 能打开这个端口的人，\n"
                "   都能用你填的密钥往对象存储写 / 删文件。公共网络（咖啡馆、宿舍、公司网）\n"
                "   请改用 --host 127.0.0.1。\n" % (args.host, args.port))
        else:
            print("生成器：http://%s:%d/dev/theme-builder/" % (args.host, args.port))
        print("上传接口：/api/s3/test /api/s3/upload /api/s3/list /api/s3/delete")
        print("Ctrl+C 停止")
        try:
            httpd.serve_forever()
        except KeyboardInterrupt:
            print("\n已停止")


if __name__ == "__main__":
    main()