#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""TamaBox 部署后冒烟巡检（端到端）。

用途：对已部署环境跑一遍核心链路，用于发版/重新部署后的回归。
覆盖：注册 -> 登录 -> 个人页/提问列表/投票面板 -> 投票全链路（建票发布/公开页/匿名提交/
      结果导出计数/草稿/表单型）-> 匿名提问 -> 停用
      -> /deactivated 成功页 -> **同邮箱重注册必须被明确拒绝并提示联系站长**
      -> 公开页只读巡检 -> /deactivated 四语言渲染。

语义约定（2026-09-15 起）：停用 = 清空内容 + 保留用户行作「身份占位」，
故邮箱/域名继续被占用，重注册应得到「账号已停用，请找站长」的提示而不是 302 或 500。

用法：
    python dev/devcheck-e2e.py [--base https://box-dev-api.337845818.best]

注意：
- 会在目标环境**真实创建账号并发送提问**（使用 tbe2e<时间戳>@example.com 之类的可识别测试邮箱），
  仅在测试/开发环境运行。
- 每次完整运行都会**留下一个已停用的身份占位**（邮箱/域名被占用）。清理方式：
  `TamaBox ban --email <该邮箱> --yes`（连行物理删除）或 `TamaBox restore --email <该邮箱> --yes`（恢复账号）。
- 依赖数学验证码（conf.Recaptcha.CaptchaType == "math"）；若站点改用 recaptcha/turnstile 则无法自动通过。
"""
import argparse
import http.cookiejar
import json
import re
import sys
import time
import urllib.error
import urllib.parse
import urllib.request

UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
      "(KHTML, like Gecko) Chrome/124.0 Safari/537.36")

RESULTS = []


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, req, fp, code, msg, headers, newurl):  # noqa: D102
        return None


class Resp:
    def __init__(self, status, headers, body):
        self.status = status
        self.h = {k.lower(): v for k, v in headers.items()}
        self.body = body

    @property
    def loc(self):
        return self.h.get("location", "")

    def alert(self):
        m = re.search(r'class="uk-alert[^"]*"[^>]*>(.*?)</div>', self.body, re.S)
        return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", m.group(0))).strip()[:200] if m else ""


class Client:
    def __init__(self, base):
        self.base = base
        self.cj = http.cookiejar.CookieJar()
        self.op = urllib.request.build_opener(
            urllib.request.HTTPCookieProcessor(self.cj), NoRedirect())

    def req(self, method, path, data=None, timeout=90):
        url = path if path.startswith("http") else self.base + path
        headers = {"User-Agent": UA, "Referer": self.base + path.split("?")[0]}
        body = urllib.parse.urlencode(data).encode() if data is not None else None
        if body:
            headers["Content-Type"] = "application/x-www-form-urlencoded"
        req = urllib.request.Request(url, data=body, headers=headers, method=method)
        try:
            with self.op.open(req, timeout=timeout) as r:
                return Resp(r.status, dict(r.headers), r.read().decode("utf-8", "replace"))
        except urllib.error.HTTPError as e:
            # 读取错误响应体本身也可能超时（尤其 302 无体/远端慢），不能让它把巡检整轮打断
            try:
                body = e.read().decode("utf-8", "replace")
            except Exception:  # noqa: BLE001
                body = ""
            return Resp(e.code, dict(e.headers), body)
        except Exception as e:  # noqa: BLE001
            return Resp(-1, {}, "TRANSPORT_ERROR: %s: %s" % (type(e).__name__, e))

    def get(self, p, **kw):
        # 只读 GET 重试一次：远端偶发慢请求会产生 TRANSPORT_ERROR，避免巡检出现假失败。
        r = self.req("GET", p, **kw)
        if r.status == -1:
            r = self.req("GET", p, **kw)
        return r

    def post(self, p, data=None, **kw):
        return self.req("POST", p, data=data, **kw)


def csrf_of(html):
    m = re.search(r'name="_csrf"\s+value="([^"]+)"', html)
    return m.group(1) if m else None


def captcha(c):
    """取一道数学验证码并算出答案，返回 <id>:<answer>。"""
    j = json.loads(c.get("/api/captcha/generate").body)
    pid, q = j["data"]["id"], j["data"]["question"]
    m = re.match(r"\s*(\d+)\s*([+\-×])\s*(\d+)\s*=", q)
    a, op, b = int(m.group(1)), m.group(2), int(m.group(3))
    ans = {"+": a + b, "-": a - b, "×": a * b}[op]
    return "%s:%d" % (pid, ans)


def step(desc, ok, extra=""):
    RESULTS.append((desc, ok, extra))
    print("  [%s] %s%s" % ("PASS" if ok else "FAIL", desc, ("  (%s)" % extra) if extra else ""))


# 停用提示的跨语言特征串（站点默认语言可能是 ja，断言不能只看中文）
DEACTIVATED_HINTS = ("已停用", "has been deactivated", "は停止されています")
INTERNAL_ERROR_HINTS = ("サーバー内部エラー", "Internal Server Error", "服务器内部错误")


def is_deactivated_rejected(r):
    """停用后的重注册应被明确拒绝：HTTP 200 + 提示「账号已停用」，且不得是 500 兜底。"""
    if r.status != 200:
        return False
    if any(h in r.body for h in INTERNAL_ERROR_HINTS):
        return False
    return any(h in r.body for h in DEACTIVATED_HINTS)


def check_public_pages(base):
    """公开页只读巡检：期望全部 200（/user/votes 之类的登录页除外）。"""
    paths = ["/", "/explore", "/register", "/login", "/forgot-password",
             "/sponsor", "/change-logs", "/about-me", "/robots.txt",
             "/sitemap.xml", "/llms.txt", "/llms-full.txt", "/favicon.ico",
             "/deactivated", "/api/captcha/generate", "/api/v1/site-stats"]
    c = Client(base)
    bad = []
    for p in paths:
        r = c.get(p, timeout=40)
        if r.status != 200:
            bad.append("%s=%s" % (p, r.status))
    step("公开页只读巡检（%d 个端点）" % len(paths), not bad, ", ".join(bad))


def create_vote(c, title, vtype="single", options=None, fields=None, action="publish"):
    """提交一次投票创建（POST /user/votes）。

    后端对创建动作一律按 draft=true 解析（草稿可随时暂存、跳过审核），
    action=publish 时再走 validateForPublish + 内容审核；因此这里两种动作都可直接测。
    """
    page = c.get("/user/votes/new")
    data = {
        "title": title, "description": "", "visibility": "public",
        "voter_scope": "anyone", "show_result": "after_vote",
        "type": vtype, "max_choices": "1", "allow_custom": "",
        "start_at": "", "end_at": "", "quota_limit": "0",
        "options": json.dumps(options or [], ensure_ascii=False),
        "form_fields": json.dumps(fields or [], ensure_ascii=False),
        "action": action, "_csrf": csrf_of(page.body),
    }
    return c.post("/user/votes", data)


def check_user_assets(owner):
    """登录态的接口/导出，以及停用页的「身份占位」说明。

    停用页说明用**结构断言**（.uk-text-muted 段落非空），不写死文案 —— 站点语言随会话变化。
    """
    r = owner.get("/api/v1/user")
    step("登录态 /api/v1/user 返回 JSON", r.status == 200 and r.body.lstrip().startswith("{"),
         "HTTP %s" % r.status)
    step("登录态 /explore", owner.get("/explore").status == 200)

    csrf = csrf_of(owner.get("/user/profile").body)
    r = owner.post("/user/profile/export-json", {"_csrf": csrf})
    step("导出 JSON", r.status == 200 and r.body.lstrip().startswith(("{", "[")),
         "HTTP %s" % r.status)
    r = owner.post("/user/profile/export", {"_csrf": csrf})
    step("导出 Excel", r.status == 200, "HTTP %s %s" % (r.status, r.h.get("content-type", "")))

    r = owner.get("/user/profile/deactivate")
    note = re.search(r'class="uk-text-small uk-text-muted">(.*?)</span>', r.body, re.S)
    step("停用页含身份占位说明", r.status == 200 and bool(note) and note.group(1).strip() != "")


def check_vote_flow(owner, base):
    """投票全链路：建票发布 -> 管理列表 -> 公开页 -> 匿名提交 -> 导出计数 -> 草稿 -> 表单型。"""
    ts = time.strftime("%m%d%H%M%S")

    title = "巡检单选投票%s" % ts
    r = create_vote(owner, title, "single", options=["选项甲", "选项乙", "选项丙"])
    step("创建单选投票并发布", r.status == 302 and "questions" in r.loc,
         "HTTP %s -> %s %s" % (r.status, r.loc, r.alert()))

    q = owner.get("/user/questions")
    step("投票出现在管理列表", title in q.body)
    mt = re.search(r"copyShareURL\('([^']+)'", q.body)
    mv = re.search(r"/user/votes/(\d+)/edit", q.body)
    token, vid = (mt.group(1) if mt else None), (mv.group(1) if mv else None)
    step("取得分享 token / 投票 ID", bool(token and vid), "%s / %s" % (token, vid))
    if not (token and vid):
        return

    anon = Client(base)
    pv = anon.get("/vote/%s" % token)
    step("公开投票页渲染", pv.status == 200 and title in pv.body, "HTTP %s" % pv.status)
    mo = re.search(r'name="option_id" value="(\d+)"', pv.body)
    step("公开页含可选项", bool(mo))
    if not mo:
        return

    sub = anon.post("/vote/%s/submit" % token, {
        "option_id": mo.group(1), "g-recaptcha-response": captcha(anon),
        "custom_option": "", "_csrf": csrf_of(pv.body),
    })
    step("匿名提交投票", sub.status == 302, "HTTP %s -> %s %s" % (sub.status, sub.loc, sub.alert()))

    # 导出仅作只读校验：单选导出的 CSV 每行是 <选项,票数>，末行为 <总参与,N>。
    # 只认「总参与」这一行，避免把各选项票数一起求和导致 2 倍计数。
    exp = owner.get("/user/votes/%s/export" % vid)
    mtot = re.search(r"总参与,(\d+)", exp.body)
    total = int(mtot.group(1)) if mtot else -1
    step("导出显示总参与 = 1", exp.status == 200 and total == 1,
         "HTTP %s 总参与=%s" % (exp.status, total))

    dr = create_vote(owner, "巡检草稿%s" % ts, "single", options=["甲", "乙"], action="save")
    step("保存草稿跳编辑页(?saved=1)",
         dr.status == 302 and "/user/votes/" in dr.loc and "saved=1" in dr.loc,
         "HTTP %s -> %s" % (dr.status, dr.loc))

    fr = create_vote(owner, "巡检表单投票%s" % ts, "form", fields=[
        {"field_type": "text", "label": "昵称", "required": True, "max_length": 50},
        {"field_type": "textarea", "label": "建议", "required": False, "max_length": 200},
    ])
    step("创建表单型投票并发布", fr.status == 302 and "questions" in fr.loc,
         "HTTP %s -> %s %s" % (fr.status, fr.loc, fr.alert()))


def check_deactivated_i18n(base):
    """停用成功页四语言渲染（用语言无关的 ratio:3 标记 + 标题非空）。"""
    bad = []
    for lang in ["zh-CN", "zh-TW", "en", "ja"]:
        c = Client(base)
        c.get("/i18n/%s?to=/deactivated" % lang, timeout=40)  # 302，语言写入 session
        r = c.get("/deactivated", timeout=40)                 # 再请求一次才渲染新语言
        title = re.search(r'uk-card-body uk-text-center".*?<h3[^>]*>([^<]*)</h3>',
                          r.body, re.S)
        if not title or not title.group(1).strip():
            bad.append(lang)
    step("停用成功页四语言渲染", not bad, "缺失: " + ",".join(bad) if bad else "")


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default="https://box-dev-api.337845818.best")
    ap.add_argument("--reuse-email", default="",
                    help="只跑一项：用该既有邮箱重新注册（用于复测停用后能否重注册）")
    args = ap.parse_args()
    base = args.base.rstrip("/")
    pwd = "TbCheck!2026"

    if args.reuse_email:
        domain = args.reuse_email.split("@")[0]
        c = Client(base)
        r = c.get("/register")
        r = c.post("/register", {
            "email": args.reuse_email, "domain": domain, "name": "TBCheck",
            "password": pwd, "repeat_password": pwd,
            "g-recaptcha-response": captcha(c), "language": "zh-CN", "agree_rules": "1",
            "_csrf": csrf_of(r.body),
        })
        step("以既有邮箱重新注册（已停用则应被明确拒绝并提示找站长）", is_deactivated_rejected(r),
             "HTTP %s %s" % (r.status, r.alert()))
        return 0 if all(x[1] for x in RESULTS) else 1

    ts = time.strftime("%m%d%H%M%S")
    email, domain = "tbe2e%s@example.com" % ts, "tbe2e%s" % ts
    print("== 新账号：%s / %s ==" % (email, domain))

    pub = Client(base)
    r = pub.get("/register")
    reg = pub.post("/register", {
        "email": email, "domain": domain, "name": "TB%s" % ts,
        "password": pwd, "repeat_password": pwd,
        "g-recaptcha-response": captcha(pub), "language": "zh-CN", "agree_rules": "1",
        "_csrf": csrf_of(r.body),
    })
    step("注册", reg.status == 302, "HTTP %s -> %s %s" % (reg.status, reg.loc, reg.alert()))

    owner = Client(base)
    r = owner.get("/login")
    r = owner.post("/login", {
        "email": email, "password": pwd,
        "g-recaptcha-response": captcha(owner), "_csrf": csrf_of(r.body),
    })
    step("登录", r.status == 302, "HTTP %s -> %s" % (r.status, r.loc))

    step("登录态 /user/profile", owner.get("/user/profile").status == 200)
    step("登录态 /user/questions", owner.get("/user/questions").status == 200)
    r = owner.get("/user/votes")
    step("/user/votes 302 -> /user/questions#votes",
         r.status == 302 and "questions" in r.loc, "HTTP %s -> %s" % (r.status, r.loc))
    step("新建投票页", owner.get("/user/votes/new").status == 200)
    check_user_assets(owner)
    check_vote_flow(owner, base)

    # 匿名提问：无邮箱时成功后直接 200 渲染提问成功页（不重定向），故只断言提问详情可达
    r = pub.get("/_/%s" % domain)
    step("打开提问箱", r.status == 200)
    r = pub.post("/_/%s" % domain, {
        "content": "自动化巡检测试提问（可忽略）",
        "g-recaptcha-response": captcha(pub),
        "receive_reply_email": "", "private": "", "ask_password": "",
        "start_chat_val": "", "_csrf": csrf_of(r.body),
    })
    step("提交提问", r.status in (200, 302), "HTTP %s -> %s" % (r.status, r.loc))

    r = owner.get("/user/profile/deactivate")
    step("打开停用页", r.status == 200)
    r = owner.post("/user/profile/deactivate", {"_csrf": csrf_of(r.body)})
    step("停用后跳 /deactivated", r.status == 302 and "deactivated" in r.loc,
         "HTTP %s -> %s" % (r.status, r.loc))
    r = owner.get("/deactivated")
    step("停用成功页渲染", r.status == 200 and "ratio: 3" in r.body, "HTTP %s" % r.status)

    pub2 = Client(base)
    r = pub2.get("/register")
    r = pub2.post("/register", {
        "email": email, "domain": domain, "name": "TB%sb" % ts,
        "password": pwd, "repeat_password": pwd,
        "g-recaptcha-response": captcha(pub2), "language": "zh-CN", "agree_rules": "1",
        "_csrf": csrf_of(r.body),
    })
    step("停用后同邮箱重注册被明确拒绝并提示找站长（核心回归）", is_deactivated_rejected(r),
         "HTTP %s %s" % (r.status, r.alert()))

    check_public_pages(base)
    check_deactivated_i18n(base)

    print("\n===== 汇总 =====")
    for d, ok, ex in RESULTS:
        print("%s  %s%s" % ("PASS" if ok else "FAIL", d, ("  (%s)" % ex) if ex else ""))
    n = sum(1 for x in RESULTS if not x[1])
    print("\n共 %d 项，失败 %d 项" % (len(RESULTS), n))
    return 1 if n else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        raise
    except Exception as e:  # noqa: BLE001
        # 远端抖动等意外中断时，仍把已完成项的结论打出来，避免只剩一段 traceback
        print("\n巡检意外中断：%s: %s" % (type(e).__name__, e))
        for d, ok, ex in RESULTS:
            print("%s  %s%s" % ("PASS" if ok else "FAIL", d, ("  (%s)" % ex) if ex else ""))
        sys.exit(2)