"""Word / PDF export of a half-yearly access self-attestation campaign.

The completed campaign is the SOC2 CC6.3 / ISO A.5.15 user-access-review
evidence; this renders it as a downloadable report (the "安全素材和产出").
Reuses the assessment export's CID-font registration (``_ensure_font`` /
``_FONT_NAME``) so Chinese renders in the PDF without bundling a TTF.

Synchronous BytesIO build (report bodies are small) behind a
StreamingResponse — same shape as ``assessment_export``.
"""
from __future__ import annotations

import io
import re
import uuid
from typing import Any
from xml.sax.saxutils import escape

from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.auditor import access_certification_service as acs
from app.domains.auditor.assessment_export import _FONT_NAME, _ensure_font

_REVOCATION_ZH = {
    "none": "—", "pending": "处理中", "scheduled": "已安排回收",
    "completed": "已回收", "failed": "回收失败", "manual_required": "需人工处理",
}
_TYPE_ZH = {
    "software_seat": "软件坐席", "cloud_grant": "云访问授权",
    "user_permission": "软件权限", "sso_grant": "SSO 授权", "api_key": "API Key",
}


async def _model(db: AsyncSession, campaign_id: uuid.UUID) -> dict[str, Any] | None:
    summary = await acs.build_self_attestation_summary(
        db, certification_id=campaign_id,
    )
    if summary is None:
        return None
    c = summary["campaign"]
    # The label is admin-controlled; strip anything that could break the
    # Content-Disposition filename (CR/LF/quotes/path separators) before it
    # reaches the response header.
    stem = re.sub(
        r'[^\w.-]', "_", f"access_self_attestation_{c['label']}".replace(" ", "_"),
    )
    return {"summary": summary, "campaign": c, "filename_stem": stem}


async def build_docx(
    db: AsyncSession, campaign_id: uuid.UUID,
) -> tuple[io.BytesIO, str] | None:
    from docx import Document
    from docx.shared import Pt

    model = await _model(db, campaign_id)
    if model is None:
        return None
    s = model["summary"]
    c = model["campaign"]

    doc = Document()
    doc.add_heading(f"全员访问自查报告 — {c['label']}", level=0)
    meta = doc.add_paragraph()
    meta.add_run(f"周期:{c['started_at'] or ''} ~ {c['due_at'] or ''}\n")
    meta.add_run(f"状态:{c['status']}\n")
    meta.add_run(
        f"覆盖率:{s['coverage_pct']}%(已确认 {s['decided']} / 共 {s['total_items']})\n"
    )
    bd = s["by_decision"]
    meta.add_run(
        f"决定:确认保留 {bd.get('keep', 0)}、标记清理 {bd.get('revoke', 0)}、"
        f"待确认 {bd.get('pending', 0)}\n"
    )
    meta.add_run("控制项映射:SOC2 CC6.3 / ISO 27001 A.5.15")

    # Flagged-removal disposition queue.
    doc.add_heading("标记清理项与处置状态", level=1)
    if s["flagged_removals"]:
        cols = ["持有人", "访问类型", "访问项", "处置状态"]
        table = doc.add_table(rows=1, cols=len(cols))
        table.style = "Light Grid Accent 1"
        for j, label in enumerate(cols):
            table.rows[0].cells[j].text = label
        for f in s["flagged_removals"]:
            cells = table.add_row().cells
            cells[0].text = f.get("holder_display_name") or f.get("holder_email") or ""
            cells[1].text = _TYPE_ZH.get(f["access_type"], f["access_type"])
            cells[2].text = f.get("access_label") or ""
            cells[3].text = _REVOCATION_ZH.get(
                f["revocation_status"], f["revocation_status"])
    else:
        doc.add_paragraph("无标记清理项。")

    # Non-responders.
    doc.add_heading("未响应人员", level=1)
    if s["non_responders"]:
        cols = ["人员", "未确认项", "是否逾期"]
        table = doc.add_table(rows=1, cols=len(cols))
        table.style = "Light Grid Accent 1"
        for j, label in enumerate(cols):
            table.rows[0].cells[j].text = label
        for r in s["non_responders"]:
            cells = table.add_row().cells
            cells[0].text = r.get("holder_display_name") or r.get("holder_email") or ""
            cells[1].text = str(r["pending"])
            cells[2].text = "是" if r["overdue"] else "否"
    else:
        doc.add_paragraph("全部人员均已响应。")

    for p in doc.paragraphs:
        for run in p.runs:
            run.font.size = run.font.size or Pt(11)

    buf = io.BytesIO()
    doc.save(buf)
    buf.seek(0)
    return buf, model["filename_stem"]


async def build_pdf(
    db: AsyncSession, campaign_id: uuid.UUID,
) -> tuple[io.BytesIO, str] | None:
    from reportlab.lib import colors
    from reportlab.lib.enums import TA_LEFT
    from reportlab.lib.pagesizes import A4
    from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
    from reportlab.lib.units import mm
    from reportlab.platypus import (
        Paragraph,
        SimpleDocTemplate,
        Spacer,
        Table,
        TableStyle,
    )

    _ensure_font()
    model = await _model(db, campaign_id)
    if model is None:
        return None
    s = model["summary"]
    c = model["campaign"]

    base = getSampleStyleSheet()
    title_style = ParagraphStyle("T", parent=base["Title"], fontName=_FONT_NAME, leading=24)
    h1 = ParagraphStyle("H1", parent=base["Heading1"], fontName=_FONT_NAME, leading=18)
    body = ParagraphStyle("B", parent=base["Normal"], fontName=_FONT_NAME,
                          fontSize=10, leading=15, alignment=TA_LEFT)

    def P(text: str, style):  # noqa: N802
        return Paragraph(escape(str(text)).replace("\n", "<br/>"), style)

    bd = s["by_decision"]
    story: list[Any] = [
        P(f"全员访问自查报告 — {c['label']}", title_style), Spacer(1, 6),
        P(f"周期:{c['started_at'] or ''} ~ {c['due_at'] or ''}", body),
        P(f"状态:{c['status']}", body),
        P(f"覆盖率:{s['coverage_pct']}%(已确认 {s['decided']} / 共 {s['total_items']})", body),
        P(f"决定:确认保留 {bd.get('keep', 0)}、标记清理 {bd.get('revoke', 0)}、待确认 {bd.get('pending', 0)}", body),
        P("控制项映射:SOC2 CC6.3 / ISO 27001 A.5.15", body),
        Spacer(1, 10),
    ]

    def _table(headers: list[str], data_rows: list[list[str]], widths=None):
        data = [headers] + data_rows
        tbl = Table(data, repeatRows=1, hAlign="LEFT", colWidths=widths)
        tbl.setStyle(TableStyle([
            ("FONTNAME", (0, 0), (-1, -1), _FONT_NAME),
            ("FONTSIZE", (0, 0), (-1, -1), 9),
            ("BACKGROUND", (0, 0), (-1, 0), colors.HexColor("#4472C4")),
            ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
            ("GRID", (0, 0), (-1, -1), 0.4, colors.grey),
            ("VALIGN", (0, 0), (-1, -1), "TOP"),
        ]))
        return tbl

    story.append(P("标记清理项与处置状态", h1))
    if s["flagged_removals"]:
        rows = [[
            P(f.get("holder_display_name") or f.get("holder_email") or "", body),
            P(_TYPE_ZH.get(f["access_type"], f["access_type"]), body),
            P(f.get("access_label") or "", body),
            P(_REVOCATION_ZH.get(f["revocation_status"], f["revocation_status"]), body),
        ] for f in s["flagged_removals"]]
        story.append(_table(["持有人", "访问类型", "访问项", "处置状态"], rows,
                            [35 * mm, 25 * mm, 70 * mm, 25 * mm]))
    else:
        story.append(P("无标记清理项。", body))
    story.append(Spacer(1, 10))

    story.append(P("未响应人员", h1))
    if s["non_responders"]:
        rows = [[
            P(r.get("holder_display_name") or r.get("holder_email") or "", body),
            P(str(r["pending"]), body),
            P("是" if r["overdue"] else "否", body),
        ] for r in s["non_responders"]]
        story.append(_table(["人员", "未确认项", "是否逾期"], rows,
                            [90 * mm, 30 * mm, 30 * mm]))
    else:
        story.append(P("全部人员均已响应。", body))

    buf = io.BytesIO()
    doc = SimpleDocTemplate(
        buf, pagesize=A4, topMargin=18 * mm, bottomMargin=18 * mm,
        leftMargin=18 * mm, rightMargin=18 * mm,
        title=f"Access Self-Attestation — {c['label']}",
    )
    doc.build(story)
    buf.seek(0)
    return buf, model["filename_stem"]