Nsfc Budget

skill_info:
  name: nsfc-budget
  version: 0.1.2
  description: "基于 NSFC 标书正文或补充材料生成预算说明书 LaTeX 项目并渲染 budget.pdf（不适用：仅想了解预算原则、或 2026 青年 A/B/C 默认包干制且无需预算说明书的场景）"
  author: "Bensz Conan"
  category: writing

defaults:
  template_id: "01"
  output_dirname: "budget_output"
  intermediate_dirname: ".bensz-api/skills/nsfc-budget"
  project_type: general
  total_budget_wan:
    general: 50
    local: 50
    youth: 30
  target_chars:
    recommended_min: 800
    recommended_max: 1000
    recommended_default: 900
  per_section_max_chars: 500
  indirect_rate_hint:
    general: 0.30
    local: 0.30
    youth: 0.30
  requested_total_tolerance_wan: 1.0
  compile_runs: 2

rules:
  require_workdir: true
  project_types:
    - general
    - local
    - youth
  budget_modes:
    - budget_based
    - package_based
    - historical_budget_based
  budget_scopes:
    - direct
    - total
    - to_be_confirmed
  budget_mode_default: budget_based
  package_based_2026_project_types:
    - youth_a
    - youth_b
    - youth_c
  allowed_latex_commands:
    - linebreak
    - BudgetBold
  zero_text:
    equipment: "本项目不列支设备费。"
    business: "本项目不列支业务费。"
    labor: "本项目不列支劳务费。"
    transfer: "本项目无合作研究转拨资金。"
    other_source: "本项目无其他来源资金。"

validation:
  equipment_ratio_warning: 0.50
  require_section_text_when_amount_positive: true

output:
  latex_entry: budget.tex
  pdf_name: budget.pdf
  section_files:
    equipment: extraTex/1.1.设备费.tex
    business: extraTex/1.2.业务费.tex
    labor: extraTex/1.3.劳务费.tex
    transfer: extraTex/2.1.合作研究转拨资金.tex
    other_source: extraTex/3.1.其他来源资金.tex

template_id: "01"
name: "NSFC 预算说明书通用模板"
description: "适配设备费/业务费/劳务费/合作研究转拨资金/其他来源资金五段结构的预算说明书 LaTeX 模板"
default: true
latex_entry: "budget.tex"
pdf_name: "budget.pdf"
project_types: ["general", "local", "youth"]
section_files: {"equipment": "extraTex/1.1.设备费.tex", "business": "extraTex/1.2.业务费.tex", "labor": "extraTex/1.3.劳务费.tex", "transfer": "extraTex/2.1.合作研究转拨资金.tex", "other_source": "extraTex/3.1.其他来源资金.tex"}

%!TEX program = xelatex

\documentclass[12pt,UTF8,a4paper,fontset=none]{ctexart}

\input{extraTex/@config.tex}

\begin{document}
\pagestyle{empty}

\begin{center}
  {\BudgetTitleFont 预算说明书}
\end{center}

\vspace{-1pt}

\begin{BudgetBox}
\BudgetSectionLine{1.科学基金资助项目直接费用}
\BudgetBoldParagraph{请按照《国家自然科学基金项目申请书预算编制说明》等有关要求，按照政策相符性、目标相关性和经济合\linebreak{}理性原则，实事求是编制项目预算。填报时，每个科目应结合科研任务按支出用途进行基本测算说明。}

\BudgetParagraph{1.1 设备费（是指在项目实施过程中购置或试制专用仪器设备，对现有仪器设备进行升级改造，以及租赁外单\linebreak{}位仪器设备而发生的费用。计算类仪器设备和软件工具可在设备费科目列支。填报时，应按照设备购置费、试制改\linebreak{}造费和租赁使用费的分类，提供设备支出的必要性及基本测算说明。单价大于50万元（含50万元）的设备需补充说\linebreak{}明设备的主要性能指标、主要技术参数等内容。）}

\input{extraTex/1.1.设备费.tex}

\BudgetParagraph{1.2 业务费（是指项目实施过程中消耗的各种材料、辅助材料等低值易耗品的采购、运输、装卸、整理等费用，\linebreak{}发生的测试化验加工、燃料动力、出版/文献/信息传播/知识产权事务、会议/差旅/国际合作交流等费用，以及其他\linebreak{}相关支出。填报时，应按照支出大类进行基本测算说明。）}

\input{extraTex/1.2.业务费.tex}

\BudgetParagraph{1.3 劳务费（是指在项目实施过程中支付给参与项目研究的研究生、博士后、访问学者以及项目聘用的研究人\linebreak{}员、科研辅助人员等的劳务性费用，以及支付给临时聘请的咨询专家的费用等。填报时，应综合考量劳务费支出对\linebreak{}象所承担研究任务的必要性、投入本项目的工作时长、费用标准的合理性等因素，按照人员类别进行基本测算说明。\linebreak{}专家咨询费应按照国家有关规定执行。）}

\input{extraTex/1.3.劳务费.tex}

\BudgetSectionLine{2.直接费用中合作研究转拨资金}
\BudgetParagraph{需对合作研究单位承担的研究任务做必要说明。直接费用转拨资金需经项目申请人与参与者协商一致，并按\hspace{-0.35em}设\linebreak{}备费、业务费、劳务费三个科目做预算说明。如存在多个合作研究单位，请分单位逐一说明。}

\input{extraTex/2.1.合作研究转拨资金.tex}

\BudgetSectionLine{3.其他来源资金}
\BudgetParagraph{对其他来源资金的资金来源、资金具体开支用途做简要说明。}

\input{extraTex/3.1.其他来源资金.tex}
\end{BudgetBox}

\end{document}

\usepackage[english]{babel}
\usepackage{xcolor}
\usepackage{geometry}
\usepackage{fontspec}
\usepackage{xeCJK}
\usepackage{titlesec}
\usepackage{calc}
\usepackage{mdframed}
\usepackage{ragged2e}

\geometry{left=2.51cm,right=2.34cm,top=2.85cm,bottom=1.75cm}

\definecolor{MsBlue}{RGB}{0,112,192}

% =========================
% 可调参数（建议优先改这里）
% =========================

% 中文假粗体强度。
% 用途：控制标题、提示语等中文加粗文字的“粗细程度”。
% 调参参考：1.2=很轻，1.5=偏轻，1.8=当前默认，2.0=偏明显，2.5=较重。

\newcommand{\BudgetCJKFakeBold}{0.1}

\IfFileExists{./fonts/TimesNewRoman.ttf}{
  \setmainfont{TimesNewRoman}[Path=./fonts/,Extension=.ttf]
}{
  \IfFontExistsTF{Times New Roman}{
    \setmainfont{Times New Roman}
  }{
    \setmainfont{TeX Gyre Termes}
  }
}

\IfFileExists{./fonts/FandolFang-Regular.otf}{
  \setCJKmainfont{FandolFang-Regular}[Path=./fonts/,Extension=.otf]
}{
  \setCJKmainfont{STSong}
}

% 页面标题“预算说明书”专用中文字体族。
% 按官方样式要求，封面标题使用黑体。

\IfFileExists{./fonts/FandolHei-Regular.otf}{
  \newCJKfontfamily\budgettitlefont{FandolHei-Regular}[Path=./fonts/,Extension=.otf,AutoFakeBold=\BudgetCJKFakeBold]
}{
  \newCJKfontfamily\budgettitlefont{Heiti SC}[AutoFakeBold=\BudgetCJKFakeBold]
}

% 正文中的中文加粗字体族。
% 当前策略：正文里的强调文字不切换到黑体，继续使用与正文相同的中文字体，只额外施加假粗体。
% 这样可以避免“加粗后字体风格突变”，更贴近正文原始字形。

\IfFileExists{./fonts/FandolFang-Regular.otf}{
  \newCJKfontfamily\budgetboldfont{FandolFang-Regular}[Path=./fonts/,Extension=.otf,AutoFakeBold=\BudgetCJKFakeBold]
}{
  \newCJKfontfamily\budgetboldfont{STSong}[AutoFakeBold=\BudgetCJKFakeBold]
}

\xeCJKsetup{CJKecglue={}}

\setlength{\parindent}{0pt}
\setlength{\parskip}{0pt}
\renewcommand{\baselinestretch}{1.35}

\setcounter{secnumdepth}{0}

\titleformat{\section}[block]
  {\normalfont\budgetboldfont\bfseries\fontsize{9.03bp}{12.17bp}\selectfont}
  {}
  {0pt}
  {}

\titlespacing*{\section}{0pt}{0.45\baselineskip}{0.10\baselineskip}

\titleformat{\subsection}[block]
  {\normalfont\fontsize{9.03bp}{12.17bp}\selectfont}
  {}
  {0pt}
  {}

\titlespacing*{\subsection}{0pt}{0.28\baselineskip}{0pt}

% 正文基础字号。
% 说明：预算说明书正文及多数段落宏都会复用这一字号设置。

\newcommand{\BudgetBodyFont}{\fontsize{9.03bp}{12.17bp}\selectfont}

% 页面标题“预算说明书”的字号与字体。
% 按官方样式使用“黑体四号加粗”；这里用 14pt 对应四号。

\newcommand{\BudgetTitleFont}{\budgettitlefont\bfseries\fontsize{14pt}{16pt}\selectfont}

% 行内手动加粗工具。

\newcommand{\BudgetBold}[1]{{\budgetboldfont\bfseries #1}}

% 占位空白：用于保持模板空白区域高度。

\newcommand{\BudgetPlaceholder}[1]{\vspace*{#1}}

% 正文整体左移补偿。
% 用途：微调整个段落块的视觉起点；一般不建议频繁改。
% 调参参考：数值更负=整体更靠左；更接近 0=整体更靠右。

\newcommand{\BudgetLeftShift}{-7pt}

% 段落首行缩进。
% 当前语义：严格按“2 个中文字符宽度”计算，缩进发生在序号前面。
% 若想改成 1.5 个或 3 个中文字符，可写成 1.5\ccwd 或 3\ccwd。

\newcommand{\BudgetIndent}{2\ccwd}

% 外框内部顶部到首个一级标题之间的微调量。
% 用途：单独控制 \begin{BudgetBox} 与首个 \BudgetSectionLine 之间的间隙。
% 调参参考：更负=间隙更小、内容更靠上；更接近 0=间隙更大。

\newcommand{\BudgetBoxTopGapAdjust}{-2pt}

% 一级标题行（如“1.科学基金资助项目直接费用”）。

\newcommand{\BudgetSectionLine}[1]{%
  {\budgetboldfont\bfseries\BudgetBodyFont\noindent\hspace*{\BudgetIndent}#1\par}%
}

% 普通正文段落。
% 特点：首行缩进，续行顶格，两端对齐。

\newcommand{\BudgetParagraph}[1]{%
  {\BudgetBodyFont\leftskip=\BudgetLeftShift\relax\justifying\parindent=\BudgetIndent #1\par}%
}

% 加粗正文段落。
% 用于顶部“请按照……”这类强调说明。

\newcommand{\BudgetBoldParagraph}[1]{%
  {\budgetboldfont\bfseries\BudgetBodyFont\leftskip=\BudgetLeftShift\relax\justifying\parindent=\BudgetIndent #1\par}%
}

% 外框环境：控制边框、内边距、固定高度等。
% 若想调“框本身”的粗细或上下内边距，可优先查看这里的 mdframed 参数。

\newenvironment{BudgetBox}{%
  \begin{mdframed}[
    skipabove=0pt,
    skipbelow=0pt,
    leftmargin=2pt,
    rightmargin=2pt,
    innertopmargin=7pt,
    innerbottommargin=7pt,
    innerleftmargin=3pt,
    innerrightmargin=3pt,
    linewidth=0.8pt,
    linecolor=black
  ]
    \begin{minipage}[t][22.35cm][t]{\linewidth}
    \vspace*{\BudgetBoxTopGapAdjust}
    \BudgetBodyFont
    \justifying
    \setlength{\parindent}{\BudgetIndent}%
    \setlength{\parskip}{0pt}%
}{%
    \end{minipage}
  \end{mdframed}
}

% 在此填写“设备费”预算说明；默认保留空白以贴近官方模板版式。
\BudgetPlaceholder{2.9\baselineskip}

% 在此填写“业务费”预算说明；默认保留空白以贴近官方模板版式。
\BudgetPlaceholder{2.9\baselineskip}

% 在此填写“劳务费”预算说明；默认保留空白以贴近官方模板版式。
\BudgetPlaceholder{3.6\baselineskip}

% 在此填写“合作研究转拨资金”预算说明；默认保留空白以贴近官方模板版式。
\BudgetPlaceholder{5.2\baselineskip}

% 在此填写“其他来源资金”预算说明；默认保留空白以贴近官方模板版式。
\BudgetPlaceholder{5.0\baselineskip}

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import shutil
import sys
from datetime import datetime
from pathlib import Path

SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from runtime_utils import dump_json, load_config, resolve_output_dir, resolve_under


def parse_args() -> argparse.Namespace:
    skill_root = Path(__file__).resolve().parents[1]
    config, _warnings = load_config(skill_root)
    defaults = config.get("defaults") or {}
    rules = config.get("rules") or {}
    target_chars = defaults.get("target_chars") or {}
    parser = argparse.ArgumentParser(description="Initialize a hidden nsfc-budget run directory.")
    parser.add_argument("--workdir", required=True, help="User work directory.")
    parser.add_argument(
        "--project-type",
        choices=sorted((rules.get("project_types") or (defaults.get("total_budget_wan") or {}).keys())),
        default=str(defaults.get("project_type") or "general"),
    )
    parser.add_argument("--total-budget-wan", type=float)
    parser.add_argument("--target-chars", type=int, default=int(target_chars.get("recommended_default") or 900))
    parser.add_argument("--template-id", default=str(defaults.get("template_id") or "01"))
    parser.add_argument("--output-dirname", default=str(defaults.get("output_dirname") or "budget_output"))
    parser.add_argument("--material", action="append", default=[], help="Material file or directory to snapshot.")
    return parser.parse_args()


def ensure_directory(path: Path) -> Path:
    path.mkdir(parents=True, exist_ok=True)
    return path


def unique_path(path: Path) -> Path:
    if not path.exists():
        return path
    stem = path.stem
    suffix = path.suffix
    counter = 2
    while True:
        candidate = path.with_name(f"{stem}_{counter}{suffix}")
        if not candidate.exists():
            return candidate
        counter += 1


def validate_args(args: argparse.Namespace) -> None:
    if args.target_chars <= 0:
        raise ValueError("target_chars 必须大于 0")
    if args.total_budget_wan is not None and args.total_budget_wan < 0:
        raise ValueError("total_budget_wan 不能为负数")


def copy_material(source: Path, destination_dir: Path) -> str:
    target = unique_path(destination_dir / source.name)
    if source.is_dir():
        shutil.copytree(source, target)
    else:
        shutil.copy2(source, target)
    return str(target)


def build_spec(args: argparse.Namespace, workdir: Path, run_dir: Path, snapped_materials: list[str], config: dict) -> dict:
    defaults = config.get("defaults") or {}
    total_budget = args.total_budget_wan
    if total_budget is None:
        total_budget = float(((defaults.get("total_budget_wan") or {}).get(args.project_type)) or 0)
    indirect_rate_hint = float((((defaults.get("indirect_rate_hint") or {}).get(args.project_type)) or 0.30))
    return {
        "meta": {
            "project_title": "",
            "project_type": args.project_type,
            "budget_mode": str((config.get("rules") or {}).get("budget_mode_default") or "budget_based"),
            "budget_scope": "to_be_confirmed",
            "requested_total_wan": total_budget,
            "direct_costs_total_wan": None,
            "indirect_costs_wan": None,
            "target_chars_min": max(0, args.target_chars - 100),
            "target_chars_max": args.target_chars + 100,
            "per_section_max_chars": int(defaults.get("per_section_max_chars") or 500),
            "template_id": args.template_id,
            "workdir": str(workdir),
            "output_dirname": args.output_dirname,
            "run_dir": str(run_dir),
            "materials": snapped_materials,
            "notes": "",
        },
        "budget": {
            "equipment_wan": 0.0,
            "business_wan": 0.0,
            "labor_wan": 0.0,
            "transfer_wan": 0.0,
            "other_source_wan": 0.0,
            "requested_total_tolerance_wan": float(defaults.get("requested_total_tolerance_wan") or 1.0),
            "indirect_rate_hint": indirect_rate_hint,
        },
        "sections": {
            "equipment": {"amount_wan": 0.0, "paragraphs": []},
            "business": {"amount_wan": 0.0, "paragraphs": []},
            "labor": {"amount_wan": 0.0, "paragraphs": []},
            "transfer": {"amount_wan": 0.0, "paragraphs": []},
            "other_source": {"amount_wan": 0.0, "paragraphs": []},
        },
        "evidence": {
            "pricing_basis": [],
            "assumptions": [],
            "items_to_confirm": [],
        },
    }


def main() -> int:
    args = parse_args()
    skill_root = Path(__file__).resolve().parents[1]
    config, config_warnings = load_config(skill_root)
    defaults = config.get("defaults") or {}
    workdir = Path(args.workdir).expanduser().resolve()
    if not workdir.exists() or not workdir.is_dir():
        print(f"[nsfc-budget] workdir does not exist or is not a directory: {workdir}", file=sys.stderr)
        return 2

    try:
        validate_args(args)
        resolve_output_dir(
            workdir,
            args.output_dirname,
            str(defaults.get("intermediate_dirname") or ".bensz-api/skills/nsfc-budget"),
            label="output_dirname",
        )
        template_dir = resolve_under(skill_root / "models", args.template_id, label="template_id")
        if not template_dir.exists() or not template_dir.is_dir():
            raise ValueError(f"template_id 对应模板不存在：{args.template_id}")
    except ValueError as exc:
        print(f"[nsfc-budget] {exc}", file=sys.stderr)
        return 2

    intermediate_root = ensure_directory(workdir / str(defaults.get("intermediate_dirname") or ".bensz-api/skills/nsfc-budget"))
    run_dir = unique_path(intermediate_root / datetime.now().strftime("%Y-%m-%d-%H-%M"))
    ensure_directory(run_dir)
    ensure_directory(run_dir / "input")
    ensure_directory(run_dir / "output")
    ensure_directory(run_dir / "log")
    materials_dir = ensure_directory(run_dir / "input" / "materials")
    ensure_directory(run_dir / "logs")
    ensure_directory(run_dir / "build")

    snapped_materials = []
    for raw in args.material:
        source = Path(raw).expanduser().resolve()
        if not source.exists():
            print(f"[nsfc-budget] warning: material not found, skipped: {source}", file=sys.stderr)
            continue
        snapped_materials.append(copy_material(source, materials_dir))

    spec = build_spec(args, workdir, run_dir, snapped_materials, config)
    spec_path = run_dir / "budget_spec.json"
    dump_json(spec_path, spec)

    manifest = {
        "workdir": str(workdir),
        "intermediate_root": str(intermediate_root),
        "run_dir": str(run_dir),
        "materials_dir": str(materials_dir),
        "spec_path": str(spec_path),
        "config_warnings": config_warnings,
    }
    dump_json(run_dir / "run_manifest.json", manifest)
    (intermediate_root / "ACTIVE_RUN.txt").write_text(str(run_dir) + "\n", encoding="utf-8")

    print(json.dumps(manifest, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3

from __future__ import annotations

import argparse
import json
import re
import shutil
import subprocess
import sys
from pathlib import Path
from typing import Any

SCRIPT_DIR = Path(__file__).resolve().parent
if str(SCRIPT_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPT_DIR))

from runtime_utils import dump_json, load_config, load_template_meta, resolve_output_dir, resolve_under, safe_rel_path


SECTION_KEYS = ["equipment", "business", "labor", "transfer", "other_source"]
TEMPLATE_IGNORE = shutil.ignore_patterns(
    ".DS_Store",
    "*.aux",
    "*.log",
    "*.synctex.gz",
    "budget.pdf",
)


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Render NSFC budget project from budget_spec.json.")
    parser.add_argument("--spec", required=True, help="Path to budget_spec.json.")
    parser.add_argument("--force", action="store_true", help="Remove output directory before rendering.")
    parser.add_argument("--skip-compile", action="store_true", help="Skip xelatex compilation.")
    return parser.parse_args()


def load_json(path: Path) -> dict[str, Any]:
    return json.loads(path.read_text(encoding="utf-8"))


def save_json(path: Path, data: dict[str, Any]) -> None:
    dump_json(path, data)


def collapse_whitespace(text: str) -> str:
    return re.sub(r"\s+", " ", text).strip()


def escape_latex_text(text: str) -> str:
    replacements = {
        "\\": r"\textbackslash{}",
        "&": r"\&",
        "%": r"\%",
        "$": r"\$",
        "#": r"\#",
        "_": r"\_",
        "{": r"\{",
        "}": r"\}",
        "~": r"\textasciitilde{}",
        "^": r"\textasciicircum{}",
    }
    return "".join(replacements.get(char, char) for char in text)


def sanitize_paragraph_latex(text: str) -> str:
    tokens: dict[str, str] = {}

    def reserve(value: str) -> str:
        token = f"@@NSFCBUDGETTOKEN{len(tokens)}@@"
        tokens[token] = value
        return token

    text = re.sub(r"\\linebreak\{\}", lambda _: reserve(r"\linebreak{}"), text)

    def replace_bold(match: re.Match[str]) -> str:
        inner = escape_latex_text(match.group(1))
        return reserve(rf"\BudgetBold{{{inner}}}")

    text = re.sub(r"\\BudgetBold\{([^{}]*)\}", replace_bold, text)
    escaped = escape_latex_text(text)
    for token, value in tokens.items():
        escaped = escaped.replace(token, value)
    return escaped


def strip_allowed_latex(text: str) -> str:
    text = re.sub(r"\\linebreak\{\}", "", text)
    text = re.sub(r"\\BudgetBold\{([^{}]*)\}", r"\1", text)
    text = re.sub(r"[{}]", "", text)
    return collapse_whitespace(text)


def visible_char_count(paragraphs: list[str]) -> int:
    joined = "".join(strip_allowed_latex(item) for item in paragraphs)
    return len(re.sub(r"\s+", "", joined))


def validate_latex_commands(paragraphs: list[str], allowed_commands: set[str]) -> list[str]:
    errors = []
    for index, paragraph in enumerate(paragraphs, start=1):
        commands = re.findall(r"\\([A-Za-z]+)", paragraph)
        disallowed = sorted({item for item in commands if item not in allowed_commands})
        if disallowed:
            errors.append(f"第 {index} 段包含未允许的 LaTeX 命令：{', '.join(disallowed)}")
    return errors


def as_float(value: Any) -> float | None:
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value)
    if isinstance(value, str) and value.strip():
        return float(value.strip())
    return None


def render_paragraphs(paragraphs: list[str], fallback: str) -> str:
    effective = [collapse_whitespace(item) for item in paragraphs if collapse_whitespace(item)]
    if not effective:
        effective = [fallback]
    return "\n\n".join(f"\\BudgetParagraph{{{sanitize_paragraph_latex(item)}}}" for item in effective) + "\n"


def resolve_template(skill_root: Path, template_id: str) -> tuple[Path, dict[str, Any], list[str]]:
    template_dir = resolve_under(skill_root / "models", template_id, label="template_id")
    if not template_dir.exists():
        raise FileNotFoundError(f"template not found: {template_dir}")
    template_meta, warnings = load_template_meta(template_dir)
    if not isinstance(template_meta, dict):
        template_meta = {}
    return template_dir, template_meta, warnings


def prepare_output_dir(output_dir: Path, template_dir: Path, force: bool) -> None:
    if output_dir.exists():
        if any(output_dir.iterdir()):
            if not force:
                raise FileExistsError(f"output directory already exists and is not empty: {output_dir}")
            shutil.rmtree(output_dir)
        else:
            output_dir.rmdir()
    shutil.copytree(template_dir, output_dir, ignore=TEMPLATE_IGNORE, dirs_exist_ok=False)


def validate_spec(spec_path: Path, spec: dict[str, Any], config: dict[str, Any], skill_root: Path) -> tuple[list[str], list[str], dict[str, Any]]:
    errors = []
    warnings = []

    meta = spec.get("meta") or {}
    budget = spec.get("budget") or {}
    sections = spec.get("sections") or {}
    defaults = config.get("defaults") or {}
    rules = config.get("rules") or {}
    validation_cfg = config.get("validation") or {}

    allowed_commands = set(rules.get("allowed_latex_commands") or [])
    project_types = set(rules.get("project_types") or (defaults.get("total_budget_wan") or {}).keys())
    budget_modes = set(rules.get("budget_modes") or {"budget_based", "package_based", "historical_budget_based"})
    budget_scopes = set(rules.get("budget_scopes") or {"direct", "total", "to_be_confirmed"})

    project_type = str(meta.get("project_type") or defaults.get("project_type") or "general")
    if project_types and project_type not in project_types:
        errors.append(f"meta.project_type 非法：{project_type}")

    budget_mode = str(meta.get("budget_mode") or rules.get("budget_mode_default") or "budget_based")
    if budget_mode not in budget_modes:
        errors.append(f"meta.budget_mode 非法：{budget_mode}")

    workdir_raw = meta.get("workdir")
    if not workdir_raw:
        errors.append("meta.workdir 不能为空")
        workdir = None
    else:
        workdir = Path(str(workdir_raw)).expanduser().resolve()
        if not workdir.exists() or not workdir.is_dir():
            errors.append(f"meta.workdir 不存在或不是目录：{workdir}")

    budget_scope = str(meta.get("budget_scope") or "to_be_confirmed")
    if budget_scope not in budget_scopes:
        errors.append(f"meta.budget_scope 非法：{budget_scope}")

    output_dirname = str(meta.get("output_dirname") or defaults.get("output_dirname") or "budget_output")
    template_id = str(meta.get("template_id") or defaults.get("template_id") or "01")
    if workdir is not None:
        try:
            resolve_output_dir(
                workdir,
                output_dirname,
                str(defaults.get("intermediate_dirname") or ".bensz-api/skills/nsfc-budget"),
                label="output_dirname",
            )
        except ValueError as exc:
            errors.append(str(exc))
        intermediate_dir = workdir / str(defaults.get("intermediate_dirname") or ".bensz-api/skills/nsfc-budget")
        try:
            spec_path.resolve().relative_to(intermediate_dir.resolve())
        except Exception:
            errors.append(f"spec 必须位于 {intermediate_dir} 内：{spec_path}")
    try:
        template_dir = resolve_under(skill_root / "models", template_id, label="template_id")
        if not template_dir.exists() or not template_dir.is_dir():
            errors.append(f"template_id 对应模板不存在：{template_id}")
    except ValueError as exc:
        errors.append(str(exc))

    target_chars = defaults.get("target_chars") or {}
    target_min = int(meta.get("target_chars_min", target_chars.get("recommended_min") or 800) or 800)
    target_max = int(meta.get("target_chars_max", target_chars.get("recommended_max") or 1000) or 1000)
    per_section_max = int(meta.get("per_section_max_chars", defaults.get("per_section_max_chars") or 500) or 500)
    if target_min < 0:
        errors.append(f"meta.target_chars_min 不能为负数：{target_min}")
    if target_max <= 0:
        errors.append(f"meta.target_chars_max 必须大于 0：{target_max}")
    if target_max < target_min:
        errors.append(f"meta.target_chars_max 不能小于 target_chars_min：{target_min} > {target_max}")
    if per_section_max <= 0:
        errors.append(f"meta.per_section_max_chars 必须大于 0：{per_section_max}")

    budget_amount_keys = {
        "equipment": "equipment_wan",
        "business": "business_wan",
        "labor": "labor_wan",
        "transfer": "transfer_wan",
        "other_source": "other_source_wan",
    }

    section_char_counts = {}
    for key in SECTION_KEYS:
        section = sections.get(key) or {}
        paragraphs = section.get("paragraphs") or []
        if not isinstance(paragraphs, list):
            errors.append(f"sections.{key}.paragraphs 必须是数组")
            paragraphs = []
        command_errors = validate_latex_commands([str(item) for item in paragraphs], allowed_commands)
        errors.extend(f"sections.{key}: {item}" for item in command_errors)

        amount = as_float(section.get("amount_wan"))
        if amount is None:
            warnings.append(f"sections.{key}.amount_wan 未填写，按 0 处理")
            amount = 0.0
        if amount < 0:
            errors.append(f"sections.{key}.amount_wan 不能为负数：{amount:.2f}w")

        budget_amount = as_float(budget.get(budget_amount_keys[key]))
        if budget_amount is not None and budget_amount < 0:
            errors.append(f"budget.{budget_amount_keys[key]} 不能为负数：{budget_amount:.2f}w")
        if budget_amount is not None and abs(budget_amount - amount) > 1e-6:
            errors.append(f"budget.{budget_amount_keys[key]}={budget_amount:.2f}w 与 sections.{key}.amount_wan={amount:.2f}w 不一致")

        if bool(validation_cfg.get("require_section_text_when_amount_positive", True)) and amount > 0 and not [item for item in paragraphs if collapse_whitespace(str(item))]:
            errors.append(f"sections.{key} 金额大于 0，但正文段落为空")

        count = visible_char_count([str(item) for item in paragraphs])
        section_char_counts[key] = count
        if count > per_section_max:
            errors.append(f"sections.{key} 可见字符数 {count} 超过上限 {per_section_max}")

    equipment = as_float((sections.get("equipment") or {}).get("amount_wan")) or 0.0
    business = as_float((sections.get("business") or {}).get("amount_wan")) or 0.0
    labor = as_float((sections.get("labor") or {}).get("amount_wan")) or 0.0
    transfer = as_float((sections.get("transfer") or {}).get("amount_wan")) or 0.0
    other_source = as_float((sections.get("other_source") or {}).get("amount_wan")) or 0.0
    direct_sum = round(equipment + business + labor, 4)

    direct_costs_total = as_float(meta.get("direct_costs_total_wan"))
    requested_total = as_float(meta.get("requested_total_wan"))
    tolerance = as_float(budget.get("requested_total_tolerance_wan")) or float(defaults.get("requested_total_tolerance_wan") or 1.0)
    if direct_costs_total is not None and direct_costs_total < 0:
        errors.append(f"meta.direct_costs_total_wan 不能为负数：{direct_costs_total:.2f}w")
    if requested_total is not None and requested_total < 0:
        errors.append(f"meta.requested_total_wan 不能为负数：{requested_total:.2f}w")
    if tolerance < 0:
        errors.append(f"budget.requested_total_tolerance_wan 不能为负数：{tolerance:.2f}w")

    if direct_costs_total is not None and abs(direct_sum - direct_costs_total) > 1e-6:
        errors.append(f"设备/业务/劳务之和为 {direct_sum:.2f}w，与 meta.direct_costs_total_wan={direct_costs_total:.2f}w 不一致")

    if budget_scope == "direct" and requested_total is not None and abs(direct_sum - requested_total) > tolerance:
        errors.append(f"预算口径为 direct，但直接费用合计 {direct_sum:.2f}w 与 requested_total_wan={requested_total:.2f}w 差值超过 {tolerance:.2f}w")

    if budget_scope == "total" and direct_costs_total is None:
        warnings.append("budget_scope=total，但未填写 direct_costs_total_wan，暂无法检查总额与说明书口径的一致性")

    if transfer > direct_sum and direct_sum > 0:
        errors.append(f"合作研究转拨资金 {transfer:.2f}w 不应大于直接费用合计 {direct_sum:.2f}w")

    equipment_ratio_warning = float(validation_cfg.get("equipment_ratio_warning") or 0.50)
    if direct_sum > 0 and equipment / direct_sum > equipment_ratio_warning:
        warnings.append(f"设备费占直接费用比例约为 {equipment / direct_sum:.1%}，请核对是否符合当年政策与单位要求")

    total_chars = sum(section_char_counts.values())
    if total_chars < target_min or total_chars > target_max:
        warnings.append(f"正文总可见字符数为 {total_chars}，当前推荐区间为 {target_min}–{target_max}")

    normalized = {
        "workdir": str(workdir) if workdir else "",
        "output_dirname": output_dirname,
        "template_id": template_id,
        "section_char_counts": section_char_counts,
        "total_chars": total_chars,
        "direct_sum_wan": direct_sum,
        "transfer_wan": transfer,
        "other_source_wan": other_source,
        "budget_scope": budget_scope,
    }
    return errors, warnings, normalized


def write_sections(output_dir: Path, sections: dict[str, Any], section_files: dict[str, str], zero_text: dict[str, str]) -> None:
    for key, relative_path in section_files.items():
        section = sections.get(key) or {}
        paragraphs = [str(item) for item in (section.get("paragraphs") or [])]
        content = render_paragraphs(paragraphs, str(zero_text[key]))
        target_path = resolve_under(output_dir, relative_path, label=f"section_files.{key}")
        target_path.write_text(content, encoding="utf-8")


def compile_project(output_dir: Path, latex_entry: str, build_dir: Path, runs: int) -> None:
    try:
        for _ in range(runs):
            result = subprocess.run(
                [
                    "xelatex",
                    "-interaction=nonstopmode",
                    "-halt-on-error",
                    f"-output-directory={build_dir}",
                    latex_entry,
                ],
                cwd=output_dir,
                capture_output=True,
                text=True,
                check=False,
            )
            (build_dir / "xelatex.stdout.log").write_text(result.stdout, encoding="utf-8")
            (build_dir / "xelatex.stderr.log").write_text(result.stderr, encoding="utf-8")
            if result.returncode != 0:
                raise RuntimeError(f"xelatex failed with exit code {result.returncode}")
    except FileNotFoundError as exc:
        raise RuntimeError("xelatex 不可用，请先安装 TeX Live/MacTeX 并确保 xelatex 在 PATH 中") from exc


def write_report(run_dir: Path, spec: dict[str, Any], normalized: dict[str, Any], errors: list[str], warnings: list[str]) -> None:
    report_json = {
        "errors": errors,
        "warnings": warnings,
        "normalized": normalized,
    }
    save_json(run_dir / "validation_report.json", report_json)

    lines = [
        "# nsfc-budget 校验报告",
        "",
        f"- 模板 ID：`{normalized['template_id']}`",
        f"- 输出目录：`{normalized['output_dirname']}`",
        f"- 正文总可见字符数：`{normalized['total_chars']}`",
        f"- 直接费用合计：`{normalized['direct_sum_wan']:.2f}w`",
        "",
        "## 各部分字符数",
        "",
    ]
    for key in SECTION_KEYS:
        lines.append(f"- `{key}`：{normalized['section_char_counts'][key]}")
    lines.extend(["", "## Errors", ""])
    if errors:
        lines.extend(f"- {item}" for item in errors)
    else:
        lines.append("- 无")
    lines.extend(["", "## Warnings", ""])
    if warnings:
        lines.extend(f"- {item}" for item in warnings)
    else:
        lines.append("- 无")
    (run_dir / "validation_report.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    save_json(run_dir / "budget_spec.snapshot.json", spec)


def render_from_spec(spec_path: Path, force: bool = False, skip_compile: bool = False) -> dict[str, Any]:
    skill_root = Path(__file__).resolve().parents[1]
    config, config_warnings = load_config(skill_root)
    spec = load_json(spec_path)
    errors, warnings, normalized = validate_spec(spec_path, spec, config, skill_root)
    warnings.extend(config_warnings)
    run_dir = spec_path.parent
    template_dir = None
    template_meta: dict[str, Any] = {}
    if not errors:
        try:
            template_dir, template_meta, template_warnings = resolve_template(skill_root, normalized["template_id"])
            warnings.extend(template_warnings)
        except Exception as exc:
            errors.append(str(exc))
    write_report(run_dir, spec, normalized, errors, warnings)
    if errors:
        preview = "；".join(errors[:3])
        suffix = "；…" if len(errors) > 3 else ""
        raise ValueError(f"spec 校验失败：{preview}{suffix}；详见 {run_dir / 'validation_report.md'}")

    workdir = Path(normalized["workdir"])
    output_dir = resolve_output_dir(
        workdir,
        normalized["output_dirname"],
        str((config.get("defaults") or {}).get("intermediate_dirname") or ".bensz-api/skills/nsfc-budget"),
        label="output_dirname",
    )
    assert template_dir is not None
    prepare_output_dir(output_dir, template_dir, force=force)

    output_cfg = config.get("output") or {}
    rules_cfg = config.get("rules") or {}
    section_files = template_meta.get("section_files") or output_cfg.get("section_files") or {}
    if sorted(section_files.keys()) != sorted(SECTION_KEYS):
        raise ValueError(f"section_files 配置不完整：{section_files}")
    for key, relative_path in section_files.items():
        safe_rel_path(str(relative_path), label=f"section_files.{key}")
    zero_text = rules_cfg.get("zero_text") or {}
    if sorted(zero_text.keys()) != sorted(SECTION_KEYS):
        raise ValueError(f"zero_text 配置不完整：{zero_text}")

    write_sections(output_dir, spec.get("sections") or {}, section_files, zero_text)

    if not skip_compile:
        latex_entry = str(template_meta.get("latex_entry") or output_cfg.get("latex_entry") or "budget.tex")
        pdf_name = str(template_meta.get("pdf_name") or output_cfg.get("pdf_name") or "budget.pdf")
        safe_rel_path(latex_entry, label="latex_entry")
        safe_rel_path(pdf_name, label="pdf_name")
        build_dir = run_dir / "build"
        build_dir.mkdir(parents=True, exist_ok=True)
        compile_runs = int((config.get("defaults") or {}).get("compile_runs") or 2)
        if compile_runs <= 0:
            raise ValueError(f"compile_runs 必须大于 0：{compile_runs}")
        compile_project(output_dir, latex_entry, build_dir, runs=compile_runs)
        compiled_pdf = build_dir / pdf_name
        if not compiled_pdf.exists():
            raise FileNotFoundError(f"compiled pdf not found: {compiled_pdf}")
        shutil.copy2(compiled_pdf, output_dir / pdf_name)

    manifest = {
        "output_dir": str(output_dir),
        "pdf": str(output_dir / (str(template_meta.get("pdf_name") or output_cfg.get("pdf_name") or "budget.pdf"))) if not skip_compile else None,
        "pdf_generated": not skip_compile,
        "validation_report": str(run_dir / "validation_report.md"),
    }
    save_json(run_dir / "deliverables_manifest.json", manifest)
    return manifest


def main() -> int:
    args = parse_args()
    spec_path = Path(args.spec).expanduser().resolve()
    if not spec_path.exists():
        print(f"[nsfc-budget] spec not found: {spec_path}", file=sys.stderr)
        return 2
    try:
        manifest = render_from_spec(spec_path, force=args.force, skip_compile=args.skip_compile)
    except Exception as exc:
        print(f"[nsfc-budget] {exc}", file=sys.stderr)
        return 1
    print(json.dumps(manifest, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

#!/usr/bin/env python3

from __future__ import annotations

import json
from pathlib import Path
from typing import Any, Dict, List, Tuple


DEFAULT_CONFIG: Dict[str, Any] = {
    "defaults": {
        "template_id": "01",
        "output_dirname": "budget_output",
        "intermediate_dirname": ".bensz-api/skills/nsfc-budget",
        "project_type": "general",
        "total_budget_wan": {"general": 50, "local": 50, "youth": 30},
        "target_chars": {"recommended_min": 800, "recommended_max": 1000, "recommended_default": 900},
        "per_section_max_chars": 500,
        "indirect_rate_hint": {"general": 0.30, "local": 0.30, "youth": 0.30},
        "requested_total_tolerance_wan": 1.0,
        "compile_runs": 2,
    },
    "rules": {
        "require_workdir": True,
        "project_types": ["general", "local", "youth"],
        "budget_mode_default": "budget_based",
        "budget_modes": ["budget_based", "package_based", "historical_budget_based"],
        "budget_scopes": ["direct", "total", "to_be_confirmed"],
        "allowed_latex_commands": ["linebreak", "BudgetBold"],
        "zero_text": {
            "equipment": "本项目不列支设备费。",
            "business": "本项目不列支业务费。",
            "labor": "本项目不列支劳务费。",
            "transfer": "本项目无合作研究转拨资金。",
            "other_source": "本项目无其他来源资金。",
        },
    },
    "validation": {
        "equipment_ratio_warning": 0.50,
        "require_section_text_when_amount_positive": True,
    },
    "output": {
        "latex_entry": "budget.tex",
        "pdf_name": "budget.pdf",
        "section_files": {
            "equipment": "extraTex/1.1.设备费.tex",
            "business": "extraTex/1.2.业务费.tex",
            "labor": "extraTex/1.3.劳务费.tex",
            "transfer": "extraTex/2.1.合作研究转拨资金.tex",
            "other_source": "extraTex/3.1.其他来源资金.tex",
        },
    },
}


def merge_dict(base: Dict[str, Any], override: Dict[str, Any]) -> Dict[str, Any]:
    merged: Dict[str, Any] = dict(base)
    for key, value in override.items():
        if isinstance(value, dict) and isinstance(merged.get(key), dict):
            merged[key] = merge_dict(merged[key], value)
        else:
            merged[key] = value
    return merged


def load_yaml_file(path: Path) -> Tuple[Dict[str, Any], List[str]]:
    warnings: List[str] = []
    if not path.exists():
        return {}, [f"YAML 文件不存在：{path}"]
    try:
        import yaml  # type: ignore

        loaded = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
        if not isinstance(loaded, dict):
            warnings.append(f"YAML 解析结果不是 dict：{path}")
            return {}, warnings
        return loaded, warnings
    except ImportError:
        warnings.append("缺少 PyYAML，回退到默认配置")
    except Exception as exc:
        warnings.append(f"读取 YAML 失败，回退到默认配置：{exc}")
    return {}, warnings


def load_config(skill_root: Path) -> Tuple[Dict[str, Any], List[str]]:
    loaded, warnings = load_yaml_file(skill_root / "config.yaml")
    return merge_dict(DEFAULT_CONFIG, loaded), warnings


def load_template_meta(template_dir: Path) -> Tuple[Dict[str, Any], List[str]]:
    return load_yaml_file(template_dir / ".template.yaml")


def safe_rel_path(raw: str, *, label: str) -> Path:
    path = Path(raw)
    if not raw or raw.strip() == "":
        raise ValueError(f"{label} 不能为空")
    if path.is_absolute():
        raise ValueError(f"{label} 不能是绝对路径：{raw}")
    if any(part in {"..", "", "."} for part in path.parts):
        raise ValueError(f"{label} 不能包含越界路径段：{raw}")
    return path


def resolve_under(base: Path, relative: str, *, label: str) -> Path:
    rel = safe_rel_path(relative, label=label)
    resolved = (base / rel).resolve()
    try:
        resolved.relative_to(base.resolve())
    except ValueError as exc:
        raise ValueError(f"{label} 解析后越界：{relative} -> {resolved}") from exc
    return resolved


def paths_overlap(first: Path, second: Path) -> bool:
    first_resolved = first.resolve()
    second_resolved = second.resolve()
    try:
        first_resolved.relative_to(second_resolved)
        return True
    except ValueError:
        pass
    try:
        second_resolved.relative_to(first_resolved)
        return True
    except ValueError:
        return False


def resolve_output_dir(workdir: Path, output_dirname: str, intermediate_dirname: str, *, label: str = "output_dirname") -> Path:
    output_dir = resolve_under(workdir, output_dirname, label=label)
    workdir_resolved = workdir.resolve()
    if output_dir == workdir_resolved:
        raise ValueError(f"{label} 不能指向工作目录根路径：{output_dirname}")

    intermediate_root = resolve_under(workdir, intermediate_dirname, label="intermediate_dirname")
    if paths_overlap(output_dir, intermediate_root):
        raise ValueError(
            f"{label} 不能与隐藏工作区 {intermediate_dirname} 重叠：{output_dirname}"
        )
    return output_dir


def ensure_parent(path: Path) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)


def dump_json(path: Path, data: Dict[str, Any]) -> None:
    ensure_parent(path)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")