anafpy

<p align="center"> <a href="https://anafpy.readthedocs.io"> <img src="https://raw.githubusercontent.com/robert-malai/anafpy/main/imgs/anafpy-social-preview.png" alt="anafpy — Python 客户端，用于 ANAF e-Factura 和 e-Transport" width="720"> </a> </p> <p align="center"> <a href="https://github.com/robert-malai/anafpy/actions/workflows/ci.yml"><img src="https://img.shields.io/github/actions/workflow/status/robert-malai/anafpy/ci.yml?branch=main&label=CI" alt="CI"></a> <a href="https://codecov.io/gh/robert-malai/anafpy"><img src="https://img.shields.io/codecov/c/github/robert-malai/anafpy?branch=main" alt="Coverage"></a> <a href="https://pypi.org/project/anafpy/"><img src="https://img.shields.io/pypi/v/anafpy" alt="PyPI 版本"></a> <a href="https://pypi.org/project/anafpy/"><img src="https://img.shields.io/pypi/pyversions/anafpy" alt="Python 版本"></a> <a href="https://anafpy.readthedocs.io/en/latest/"><img src="https://app.readthedocs.org/projects/anafpy/badge/?version=latest" alt="文档"></a> <a href="https://github.com/robert-malai/anafpy/blob/main/LICENSE"><img src="https://img.shields.io/pypi/l/anafpy" alt="许可证"></a> </p> # anafpy Romania 的 **ANAF** 税务机关 Web 服务的 Python 客户端 — **e-Factura**（电子发票）、**e-Transport**（货物运输）和**公共无认证注册**（增值税/纳税人查询、财务报表）— 以及一个本地 MCP 服务器，暴露为 [Claude Cowork](https://claude.com) 技能。 anafpy 是一个**薄传输客户端** — 没有持久性，没有会计逻辑。对于 **e-Factura**，有两种方式：如果你运行了发票软件，带来它生成的发票 XML，anafpy 验证它，提交给 ANAF，跟踪状态，并拉回文件（推荐路径 — 你的系统的 XML 从不重新组合）；如果你没有，**发票创作模型** 从普通商业领域组成一个完整的 CIUS-RO 发票或信用票据 — 总额和增值税分解为你计算，根据翻译的 EN 16931 + CIUS-RO 规则集检查 — 在提交之前。无论哪种方式，请记住 **ANAF 的 SPV 不是发票存储** — 它在约 60 天后删除提交的消息 — 所以你的持久记录必须存在于你的这边：你的发票系统的分类账，或当你使用 anafpy 编写时，你下载并保存的签名 ZIP 文件。您读取的文件（您提交的发票和供应商为您开具的发票）以友好的**平面阅读视图**返回，以便轻松显示。 **e-Transport** 也完全翻译了：您从结构化字段中编写声明、UIT 删除、确认和车辆更改，无需 XML 处理，相同的模型渲染您读取的内容。 **文档： [anafpy.readthedocs.io](https://anafpy.readthedocs.io)** — 最终用户设置指南、库指南和 API 参考。 ## 你可以用它做什么？ 使用连接到 Claude 客户端的 MCP 服务器（Claude Desktop，Claude Code），罗马尼亚的会计师可以要求 Claude： **检查合作伙伴和公共数据 — 无需登录**（这些使用 ANAF 的公共，无认证服务）： - **通过 CUI/CIF 验证商业伙伴** — 名称、地址、增值税状态（plătitor de TVA）、TVA la încasare、split-VAT、inactive 标志 — 一次调用，批量调用。 - **检查合作伙伴是否已注册到 RO e-Factura。</意愿** - **查找农民注册表（RegAgric）和宗教实体注册表（RegCult）。** - **提取公司的已提交财务报表（bilanț）**，适用于给定年份。 - **验证发票 XML** 针对 ANAF 的权威服务器端 `validare`（CIUS-RO / BR-RO 规则）— 无提交。 **处理您的 e-Factura 收件箱并提交发票**（需要证书登录）： - **列出收到的和发送的发票消息**，适用于日期窗口。 - **下载发票** 作为易于阅读的视图，并 **保存官方签名 ZIP 和/或渲染的 PDF** 到磁盘 — 支持批处理流程，如“导出上个月的发票为 `<date> - <partner>.pdf`”。 - **提交发票或信用票据** — 从您的发票软件导出的 XML（推荐使用），或 **由 Claude 从普通商业领域编写** — 两种方式都是两步提交：您会看到预览，除非您明确确认，否则不会提交给 ANAF。 **在 e-Transport 中声明货物运输 — 带确认步骤**（需要登录）： - **提交声明并获取 UIT 代码**，从任何来源的运输数据 — 电子邮件、PDF 发票、CMR、电子表格 — 并 **纠正、删除、确认或更改车辆**，针对现有的 UIT。 - **列出最近的通知，检查上传状态，并查找活动声明/ UIT 代码。** - 提交是**两步确认**的：Claude 向您展示预览，除非您明确确认，否则不会提交给 ANAF。 **阅读您的 SPV 收件箱并拉取官方报告**（需要证书 — 只读）： - **检查 SPV 中收到了什么** — 收据（recipise）、付款通知、决定、通知 — 按公司和消息类型过滤，并 **保存任何文档的 PDF** 到您命名的文件夹。 - **请求官方报告并等待它们**：财政向量（VECTOR FISCAL）、未缴义务（Obligatii de plata）、申报历史（Istoric declaratii）、申报副本（D100/D112/D300/D390/D394...）、收据副本、收入证书（Adeverinte Venit）、D112↔REVISAL 不匹配 — ANAF 异步生成它们，Claude 获取 PDF。 - **查看您的证书可以查询哪些公司** — SPV 报告每次调用的授权库存。 设置注意事项：e-Factura 和 e-Transport 工具需要一次性登录您的**合格数字证书**（与您在 ANAF 的 SPV 上使用的相同）— 上述公共检查无需此证书。服务器在您自己的机器上运行**本地**，因此下载的发票和 PDF 文件保存在您自己的文件系统中。有关 **Claude Desktop + ANAF 设置**的完整设置指南，请参阅 [设置指南](https://anafpy.readthedocs.io/en/latest/mcp/setup/) — 也提供 [罗马尼亚语版本](https://anafpy.readthedocs.io/en/latest/mcp/setup.ro/)。 > 状态：**早期/alpha**（`0.x`），在 PyPI 上为 [`anafpy`](https://pypi.org/project/anafpy/)。OAuth2 身份验证层、异步客户端、双向发票创作模型和 MCP 服务器（两步提交两个服务、收件箱、下载、验证）已实现并测试，以及只读 SPV 层（证书-mTLS 收件箱：消息、下载、报告请求）。ANAF 自己的服务器端 `validare` 仍然是权威验证器；创作模型添加了一个本地翻译的规则检查，以便快速反馈。参见 [`DESIGN.md`](DESIGN.md) 以获取完整设计，[`docs/anaf-reference/`](docs/anaf-reference/) 以获取 ANAF API 的编译本地参考。 需要 **Python 3.12+**。基于 **httpx** 和 **Pydantic v2**。 ## 当前功能 - **OAuth2 身份验证层** — 授权码启动（浏览器 + 合格证书），本地令牌存储，静默刷新，通过 `anafpy` CLI 和 `httpx.Auth` 集成用于客户端。 - **`EFacturaClient`**（异步）— `upload`（准备好的 XML）和 `upload_invoice`（创作的 `InvoiceDocument`），`get_status`，`download`，`validate_signature`（检查下载发票的 MF 签名），`upload_and_wait` 轮询直到终端助手，以及 `list_messages` — 一个异步迭代器，在后台分页消息列表（按 `days` 或 `start`/`end` 窗口；空窗口 → 空迭代器，真正的 ANAF 错误 → 引发）。`download` 提供三个读取层：原始签名字节、完整的 UBL 模型和平面 `InvoiceDocument` **视图**（相同的创作模型，从线路读取完整保真度）。 - **发票创作**（`anafpy.efactura.authoring`）— 双向 CIUS-RO 模型：一个 `InvoiceDocument` 涵盖发票和信用票据，总额和增值税分解从行计算（显式值保留），以及一个翻译的 EN 16931 + CIUS-RO 规则集（`validate()`）在提交前报告发现，带有官方 BR-* id。`render_invoice` 发出上传就绪的 XML；`parse_invoice`/`read_invoice` 将线路 XML 映射回相同的模型，具有字节稳定的往返旅程。 - **`ETransportClient`**（异步）— `upload`，`get_status`，`info`，`upload_and_wait`，`list_notifications`（相同的异步迭代器形状），以及 **`upload_document`**，它组成并提交四个平面文档中的任何一个 — 一个 `FlatTransport` 声明/更正、`FlatDeletion`、`FlatConfirmation` 或 `FlatVehicleChange` — 无需调用者触摸 XML。 - **`PublicClient`**（异步）— ANAF 在 `webservicesp.anaf.ro` 上的**无身份验证**公共服务：`lookup_taxpayers`（增值税注册、增值税收款、无效、split-VAT 和 RO e-Factura 注册成员资格在一个调用中），`lookup_efactura_register`，`lookup_farmers`，`lookup_cult_entities`，`get_financial_statement`（公共 bilanț 指标）— 以及无状态的 e-Factura 文档服务：`validate_invoice`（ANAF 的权威服务器端验证，无提交）和 `render_invoice_pdf`（官方 `transformare` PDF 渲染）；两者仅在 ANAF 端生产，需要登录。无凭证，无测试/生产分割；客户端根据 ANAF 的 1 req/s 规则限制请求速率。 - **`SpvClient`**（异步，只读）— 纳税人的 **SPV 收件箱**，使用合格证书：`list_messages`（带有证书的授权库存），`download_document`（PDF），`request_report`（完整的 `cerere` 命名法 — 35 种报告类型，每种类型有类型特定的参数验证，然后再进行任何线路调用），以及 `wait_for_report`。证书步骤是交互式 `anafpy spv login`（macOS Keychain/SecureTransport 或 Windows CertStore/Schannel，通过操作系统自带的 curl — 密钥不可导出，因此 Python 的 TLS 堆栈永远不会触摸它们）；之后的所有操作都使用 APM cookie 会话，无需提示。 - **平面模型** — 上述发票创作模型同时作为收件箱阅读视图（`DownloadedMessage.view`），因此您可以下载的内容和您创作的内容具有相同的形状。e-Transport 平面模型是**双向的**：`read_flat_transport` 查看解析文档，而 `build_etransport` / `render_etransport` 编写一个 — 完整的 ANAF XSD 翻译，具有枚举编码字段（县、边境点、海关办公室、操作类型...）接受名称或代码。 - **生成模型** — UBL 2.1 / CIUS-RO（来自 `anafpy.efactura import Invoice, CreditNote`）和专有的 e-Transport XSD，从供应商模式生成。 - **MCP 服务器**（`anafpy[mcp]`）— 一个本地 stdio 连接器，将操作暴露为 Cowork 技能：先阅读，具有两个服务**两步确认提交** — e-Factura 发票（准备好的 XML 或从结构化字段组成）和 e-Transport 声明（见下文）。 同步外观已放弃作为目标 — 客户端仅异步。 ## 安装 设置一个**全新机器端到端** — ANAF 应用程序注册、证书登录和 Claude / Cowork 配置，为非开发者编写 — 遵循 [设置指南](https://anafpy.readthedocs.io/en/latest/mcp/setup/)。开发人员的简要版本： 从 [PyPI](https://pypi.org/project/anafpy/)： ```bash pip install anafpy # 或：uv add anafpy pip install 'anafpy[mcp]' # 带有 MCP 服务器 ``` 分发版提供了一个额外的功能：`anafpy[mcp]`（MCP 服务器）。 对于 MCP 服务器，请从**检出**运行（因为设置指南）：编译的 ANAF 参考资料（`docs/anaf-reference/`，作为 MCP 资源）和工作流技能（`skills/`，作为 MCP 提示）存在于存储库中，而不在轮子中。PyPI 安装的服务器运行正常，但除非 `ANAFPY_DOCS_DIR` / `ANAFPY_SKILLS_DIR` 指向副本，否则不会提供两者。从源代码： ```bash git clone https://github.com/robert-malai/anafpy && cd anafpy uv sync --all-extras ``` ## 身份验证 ANAF 使用 OAuth2（授权码）门控 **合格数字证书**。一次性交互式启动在您的机器上运行（证书存储在那里）： ```bash anafpy auth login --client-id <ID> --client-secret <SECRET> \ --redirect-uri https://localhost:9002/callback --paste anafpy auth status # 显示存储的令牌有效性 anafpy auth logout # 删除存储的令牌（注销此机器） ``` 这会打开您的浏览器进行证书步骤，捕获授权代码（粘贴或通过本地 TLS 监听器），交换令牌并存储在 **OS 凭证存储**（macOS 钥匙串，Windows 凭证管理器，Linux 秘密服务/KWallet — 默认后端；JSON 文件后端是 Docker/无头主机退出）中。然后令牌会**无头**刷新约一年（访问令牌 90 天，刷新令牌 365 天），因此证书仅需约一年一次。[身份验证指南](https://anafpy.readthedocs.io/en/latest/library/auth/)涵盖了捕获模式、令牌存储和注销。 ## 用法 客户端是异步的，作为上下文管理器使用。构建一个 `TokenProvider` 来存储您的令牌，然后调用离散操作： ```python from anafpy.auth import KeyringTokenStore, TokenProvider from anafpy.efactura import EFacturaClient provider = TokenProvider( client_id="<ID>", client_secret="<SECRET>", store=KeyringTokenStore(), # OS 凭证存储（默认后端） # 或 FileTokenStore("~/.anafpy/tokens.json") 用于无头/Docker 主机 ) async with EFacturaClient(provider) as efactura: result = await efactura.upload(invoice_xml, cif="RO12345678") status = await efactura.get_status(result.upload_id) # 或，在一个调用中：status = await efactura.upload_and_wait(invoice_xml, cif=...)``` 离散方法进行单个调用（无传输重试）。HTTP/身份验证问题引发 `AnafError` 子类；**业务结果**（`nok` 拒绝，BR-RO 发现）以类型化值返回，而不是异常。对于 HTTP 429，客户端引发 `AnafRateLimitError` 暴露 `retry_after` 而不是自退。 e-Transport 声明来自类型化模型 — 没有 XML。在 sight： ```python from anafpy.etransport import ETransportClient, FlatDeletion, FlatTransport declaration = FlatTransport(...) # 完整的示例见 e-Transport 指南 async with ETransportClient(provider) as etransport: result = await etransport.upload_document(declaration, cif="12345678") print(result.uit) # UIT 代码，在上传时发出 # 稍后：删除 / 确认 / 更改车辆到 UIT，使用相同的方式，例如 await etransport.upload_document(FlatDeletion(uit=result.uit), cif="12345678") ``` [ e-Transport 指南](https://anafpy.readthedocs.io/en/latest/library/etransport/) 提供了完整的示例。 公共注册表不需要任何身份验证： ```python from anafpy.public import PublicClient async with PublicClient() as public: lookup = await public.lookup_taxpayers(["RO12345678"]) if lookup.found: record = lookup.found[0] print(record.name, record.vat_registered, record.efactura_registered) ``` 阅读 SPV 需要一次交互登录（`anafpy spv certs` → `anafpy spv select <thumbprint>` → `anafpy spv login`，它会触发您的令牌 PIN / 2FA 一次），然后：```python from anafpy.spv import ( FileSessionStore, ReportRequest, ReportType, SpvClient, SpvSessionProvider, ) async with SpvClient(SpvSessionProvider(store=FileSessionStore())) as spv: inbox = await spv.list_messages(30) print(inbox.authorized_cuis) # CUI/CNP 您的证书可查询 result = await spv.request_report( ReportRequest(type_=ReportType.VECTOR_FISCAL, cui="12345678") ) report = await spv.wait_for_report(result.request_id) # PDF 字节 ``` ## MCP 服务器 `anafpy[mcp]` 额外功能提供了一个**本地标准输入输出 MCP 服务器**，它将客户端作为 Cowork 技能进行封装。服务器是**尽力而为**的：配置它（包括在 ANAF 门户上注册您自己的 OAuth 应用程序）是您的责任，[设置指南](https://anafpy.readthedocs.io/en/latest/mcp/setup/) 将指导您完成每一步。在运行 `anafpy auth login` 的主机上运行服务器： ```bash ANAFPY_CLIENT_ID=... ANAFPY_CLIENT_SECRET=... ANAFPY_CIF=... \ python -m anafpy.mcp # 或 `anafpy-mcp` 控制台脚本 ``` 服务器表面是**先读后写**的：自由可调用的读取工具（`anaf_*` 公共查找和 `efactura_validate` **无需登录**；`auth_status`，电子发票收件箱 - 也可以将签名的 ZIP 和 ANAF 的官方 PDF 渲染保存到磁盘 - 以及电子运输读取需要一次性登录）以及**两步确认的文件提交流程**，适用于两项服务：每个 `*_prepare*` 工具返回预览和与确切文档和 CIF 绑定的单次使用确认令牌，而匹配的 `*_submit` 文件仅在具有该令牌和 `confirm=True` 时才有效。电子发票文件可以作为您的发票软件生成的 XML（`efactura_prepare`，推荐路径）或从结构化字段组合而成（`efactura_prepare_invoice` - 无需发票软件）；电子运输申报由 `etransport_prepare*` 从结构化字段组合而成。`spv_*` 工具通过 `anafpy spv login` 建立的证书会话读取 SPV 邮箱（列出、下载 PDF 到磁盘、请求官方报告并等待其交付）- 只读，无提交。编译的 ANAF 参考资料作为只读资源提供，工作流程剧本作为 MCP 提示 - `etransport-declare`，它接受来自任何来源（电子邮件、PDF 发票、CMR）的申报，通过提取 → 准备 → 您的批准 → 提交 → UIT，以及 `personal-income-summary`，它从 SPV 收入证书中提取人员的已实现收入到每年的摘要中（带有可选的 Excel 工作簿）。请参阅 [工具概览](https://anafpy.readthedocs.io/en/latest/mcp/tools/) 和 [工作流程技能](https://anafpy.readthedocs.io/en/latest/mcp/skills/) 以获取完整的图片。 使用任何 MCP 客户端注册服务器 - 例如，使用 Claude Code，从源代码 checkout（锁定依赖，无需 PyPI）： ```bash claude mcp add anafpy \ -e ANAFPY_CLIENT_ID=... -e ANAFPY_CLIENT_SECRET=... -e ANAFPY_CIF=... \ -- uv run --directory /path/to/anafpy --frozen --extra mcp anafpy-mcp ``` **还没有凭证？** 服务器仍然启动；公共 `anaf_*` 查找（注册、财务报表）和 `efactura_validate` 完全可用。剩余的电子发票/电子运输工具在您设置凭证并在终端中运行一次 `anafpy auth login` 后解锁。 ## 开发 ```bash uv sync --all-extras uv run pytest # respx-mocked，无需凭证 uv run ruff check . && uv run ruff format --check . uv run mypy # 严格 ANAFPY_LIVE=1 uv run pytest -m live # 可选：针对真实 ANAF 进行实时测试 ``` `live` 标记重新确认与真实 ANAF 端点的线路形状，默认情况下会被跳过（不是门控）。它涵盖公共服务（无需凭证）以及使用 `.env` 凭证和 `anafpy auth login` 令牌存储的已认证 **测试** 环境：只读形状检查和两次往返，实际提交测试文档 - 最小化 CIUS-RO 发票（电子发票）和通过扁平创作模型组合的国内运输申报（电子运输）。往返仅针对测试环境，从不针对生产环境。 `efactura/ubl/` 和 `etransport/schema/` 下的模型是**生成的**（通过 `scripts/generate_*.py` 从 `schemas/` 中的已售出 XSD 生成），不允许手动编辑。请参阅 [`CLAUDE.md`](CLAUDE.md) 以获取存储库约定。 ## 许可证 [Apache-2.0](LICENSE)。独立/非官方 - 不隶属于 ANAF。 anafpy 免费使用，提供**原样**，无任何保证：它将文档移入和移出 ANAF，不提供税务建议，文件提交结果由您负责。MCP 服务器是**尽力而为**的 - 配置它，在 ANAF 门户上预配您自己的 OAuth 应用程序以及持有合格证书都由您负责（[设置指南](https://anafpy.readthedocs.io/en/latest/mcp/setup/) 涵盖所有这些）。