sdpy-kit

SDPY 统一工具集（Kit 体系），供 Python 项目开箱即用：

• 统一数据库访问层：PostgreSQL / MySQL / SQLite，三个 Kit API 完全一致，切换数据库只改一行 import；
• AI 多模型服务池：chat / embedding / rerank / image 四类能力，多厂商候选链 + 故障自动转移；
• MCP 客户端：streamable-http / sse 双传输，档案驱动多服务器；
• 工具集：Redis / 文件 / 雪花 ID / 文档互转 / Word 模板填充 / MinerU 解析 / Marker 本地解析（md + images） / 向量库 / 微信公众号文章抓取存储。

要求 Python 3.13+

模块与方法入口

安装

pip install sdpy-kit            # 基础版：config/log + kit.db 核心（SQLite 常规操作零额外依赖）
# 或
uv add sdpy-kit

可选能力按 extras 按需安装（均为延迟导入，不装不影响基础使用，缺依赖时报错会提示安装命令）：

配置（.env）

kit 通过 pydantic-settings 读取项目根目录 .env；同名真实环境变量优先于 .env。全部变量都有默认值，按需覆盖即可。

最小样例 .env

# 日志（可选）
LOG_LEVEL=INFO
LOG_TO_FILE=false

# 数据库（三选一，切换只改 DEFAULT_DB_TYPE：postgresql / mysql / sqlite）
DEFAULT_DB_TYPE=sqlite
SQLITE_DB_PATH=./data/sqlite.db
# DB_HOST=localhost
# DB_PORT=5432
# DB_DATABASE=postgres
# DB_USER=admin
# DB_PASSWORD=
# MYSQL_PORT=3306
# MYSQL_DATABASE=mydb
# MYSQL_USER=root
# MYSQL_CHARSET=utf8mb4

# 按需启用以下能力
# REDIS_HOST=localhost
# REDIS_PORT=6379
# REDIS_DB=0
# REDIS_PASSWORD=
# FILE_STORAGE_ROOT=./data
# ZVEC_DATA_DIR=./data/zvec

# AI：档案路径 + 你为各模型档案自定的密钥变量名（变量名在 ai_models.json 里声明）
# AI_MODELS_CONFIG_PATH=./config/ai_models.json
# SILICONFLOW_API_KEY=sk-xxxxxx

# MCP：档案路径 + 你为各服务器自定的 token 变量名（变量名在 mcp_servers.json 里声明）
# MCP_SERVERS_CONFIG_PATH=./config/mcp_servers.json
# AMINER_TOKEN=xxxxxx

# MinerU（仅精准解析需要 Key，Agent 轻量解析免登录）
# MINERU_API_KEY=

# Marker（本地解析，可选依赖 sdpy-kit[marker]）
# MARKER_MODE=fast              # None=按设备自动（GPU→balanced / CPU、MPS→fast）
# MARKER_HF_ENDPOINT=https://hf-mirror.com
# MARKER_HF_DISABLE_XET=true
# MARKER_LLAMA_CPP_BINARY=

环境变量 ↔ 能力对照表

表中未出现的模块（kit.common_kit、kit.docx_kit、kit.md2doc_kit、kit.wx_article_kit）不读取任何环境变量。

基础与日志（影响 kit.config 及所有模块的日志输出）

数据库 → kit.db、SQLiteKit / PgKit / MySqlKit

Redis → kit.redis_kit

文件存储 → kit.file_kit

向量库 → kit.zvec_kit

AI → kit.ai

MCP → kit.mcp_kit

MinerU → kit.mineru_kit

Marker → kit.marker_kit

这四项由 kit 在导入 marker 之前经 os.environ.setdefault 应用：仅当用户未显式设置对应环境变量时生效，不覆盖显式配置。

云 OCR（任何 ocr="hosted" 的转换）→ kit.anydoc_kit

环境变量的四类来源

密钥解析优先级

AI 的 load_dotenv(override=False) 表示已存在的真实环境变量不会被 .env 覆盖——生产用容器 / 系统环境变量注入时，可以放心保留开发用的 .env。

改了配置何时生效

kit.config —— 配置与日志

from kit.config import get_settings, get_default_logger

settings = get_settings()               # Settings 单例（lru_cache，读取 .env）
logger = get_default_logger(__name__)   # 统一日志入口（禁止自建 logging handler）
logger.info("hello %s", settings.FILE_STORAGE_ROOT)

kit.db —— 统一数据库访问层

kit/__init__.py 直接重导出 kit.db 的全部公开符号，下列两种写法等价：

from kit.db import session_scope, transactional, DBError    # 推荐
from kit import session_scope, transactional, DBError       # 等价

1. 三个数据库 Kit（24 个共享方法 API 完全一致）

共有方法（23 个，全部为 classmethod，内部自动提交）

列表参数（IN）：query / execute 的 params 中值为 list/tuple/set/frozenset 时按 IN 展开，可直接写 WHERE id IN :ids（三库通用）。IN 不保证顺序；参数列表过大 应分批（PG 单语句上限 65535 参数）。

LIKE 的方言差异：SQLite 对 ASCII 不区分大小写，PostgreSQL 区分（用 ILIKE）； 跨库一致写法：LOWER(col) LIKE LOWER(:pattern)。

导出面：from kit.db import PgKit, SQLiteKit, build_pg_url, to_vector_literal, … （kit 顶层与 kit.db 导出同一套符号）。

from sqlalchemy import BigInteger
from sqlmodel import SQLModel, Field
from kit.common_kit import CommonKit
from kit.db import SQLiteKit          # 也可 from kit.db.sqlite_kit import SQLiteKit

class User(SQLModel, table=True):
    # 主键：64 位整数 + 雪花 ID（应用侧生成，不用数据库自增）
    id: int = Field(default_factory=CommonKit.generate_snowflake_id,
                    primary_key=True, sa_type=BigInteger)
    name: str

SQLiteKit.connect()
SQLiteKit.create_tables([User])

user = SQLiteKit.create(User, {"name": "张三"})           # 自动提交
SQLiteKit.update(User, user.id, {"name": "李四"})
users = SQLiteKit.list(User, skip=0, limit=10, order_by="id", desc=True)

rows = SQLiteKit.query("SELECT * FROM user WHERE name = :name", {"name": "李四"})
affected = SQLiteKit.execute("UPDATE user SET name = '王五' WHERE id = :id", {"id": user.id})
rows = SQLiteKit.query("SELECT * FROM user WHERE id IN :ids", {"ids": [1, 2, 3]})   # 列表参数

with SQLiteKit.transaction() as session:                  # 多操作原子提交
    SQLiteKit.create(User, {"name": "批量一"})
    SQLiteKit.create(User, {"name": "批量二"})

切换数据库：把 import 的 SQLiteKit 换成 PgKit / MySqlKit 即可，其余代码不变。 注意三点例外：① create_raw_table 的列定义文本须按目标库方言改写；② 向量能力 三库同名同参但能力不同（可由 VECTOR_CAPABILITIES feature-detect），尤其 MySQL 侧检索不可用；③ create_vector_table 的 dims 为必填参数。

2. 向量方法（三库同名同参，能力差异化）

签名在 SQLiteKit / PgKit / MySqlKit 上完全一致；能力按后端不同（见下表）。 extra 前提：SQLite 需 sdpy-kit[sqliteai]；PG 需 sdpy-kit[pg] 且服务端已装 pgvector （create_vector_table 会自动尝试 CREATE EXTENSION）；MySQL 需 9.x（Community 即可存， 但检索需 HeatWave on OCI / MySQL AI）——建表前会校验服务端版本，低于 9.x 抛 DBError。

id_col / column（逃生口）：让自建 DDL 的向量表也能读写/检索。需要复合主键、额外约束、 分区或与既有业务表合表时，用 execute() 自建 DDL 再用这两个参数接入；SQLite 侧要求 id_col 为 rowid 别名（INTEGER PRIMARY KEY）。

exact 参数（默认 True = 禁止近似加速）：

能力差异（仅 L2 可跨库比较数值；float32 vs float64，需带容差）：

MySQL DISTANCE() / VECTOR_DISTANCE() 仅 HeatWave on OCI 与 MySQL AI 提供 （Community 实测报 1305 FUNCTION ... does not exist）；故 MySqlKit 只覆盖存储侧。

主键与 id（三库统一）：向量表主键均为 BIGINT（SQLite 用等价的 64 位 INTEGER PRIMARY KEY）， 不使用数据库自增；ids=None 时由 Kit 用 CommonKit.generate_snowflake_id() 生成雪花 ID （18~19 位，int64）。 ⚠ 雪花 ID 超过 2^53，经 JSON 传给 JavaScript 会丢精度，需按字符串传输； 多实例部署需各自 CommonKit.configure(datacenter_id, machine_id)（默认 (1,1) 会重复）。

# SQLite（默认精确：distance 为真值 L2）
from kit.db.sqlite_kit import SQLiteKit
SQLiteKit.create_vector_table("images", 384)          # dims 必填
SQLiteKit.insert_vectors("images", [[0.1] * 384, [0.2] * 384])
rows = SQLiteKit.search_vectors("images", [0.1] * 384, top_k=5)
rows = SQLiteKit.search_vectors("images", [0.1] * 384, top_k=5, exact=False)  # 旧量化口径（不提速）

# PostgreSQL（签名相同；可选 COSINE/L2/L1/DOT/SQUARED_L2）
from kit.db.pg_kit import PgKit
PgKit.create_vector_table("images", 384)
PgKit.insert_vectors("images", [[0.1] * 384, [0.2] * 384])
rows = PgKit.search_vectors("images", [0.1] * 384, top_k=5, distance_type="COSINE")
# 大表 + 自建索引时才用 exact=False（否则索引不被使用，见上表）
PgKit.execute("CREATE INDEX ON images USING hnsw (embedding vector_cosine_ops)")
rows = PgKit.search_vectors("images", [0.1] * 384, top_k=5,
                            distance_type="COSINE", exact=False)

# MySQL（仅存储；search_vectors 抛 VectorNotSupportedError）
from kit.db.mysql_kit import MySqlKit
MySqlKit.create_vector_table("images", 384)
MySqlKit.insert_vectors("images", [[0.1] * 384, [0.2] * 384])

# 能力自省（无需 try/except）
PgKit.VECTOR_CAPABILITIES
# {'store': True, 'search': True, 'exact': True,
#  'distances': ('COSINE', 'L2', 'L1', 'DOT', 'SQUARED_L2'),
#  'distance_kind': 'true', 'fast_distance_kind': 'true'}

# 自建 DDL 的表：用 id_col / column 接入
PgKit.execute("CREATE TABLE my_vec (vid BIGINT PRIMARY KEY, vec vector(384))")
PgKit.insert_vectors("my_vec", [[0.1] * 384], ids=[1], id_col="vid", column="vec")
PgKit.search_vectors("my_vec", [0.1] * 384, top_k=5, id_col="vid", column="vec")

0.2.2 行为变更（升级须知）

新增（非破坏性）：VECTOR_CAPABILITIES 能力声明、id_col / column 逃生口参数、 insert_vectors 支持二维 numpy 数组与批量提交。

0.2.4 变更

0.2.3 变更

两段查询范式（推荐用法）：search_vectors 只返回 [{"id", "distance"}]，载荷二次取回：

hits = SQLiteKit.search_vectors("item_vectors", query_vector, top_k=5)
rows = SQLiteKit.query("SELECT id, title FROM items WHERE id IN :ids",
                       {"ids": [h["id"] for h in hits]})   # IN 不保证顺序，需按 hits 重排

自有事务内写向量（推荐：build_insert_vectors + transaction()）：

from sqlalchemy import text

with SQLiteKit.transaction() as session:          # 换成 PgKit / MySqlKit 写法相同
    session.execute(text("INSERT INTO docs_item (id, title) VALUES (:id, :t)"),
                    {"id": doc_id, "t": title})
    for sql, params in SQLiteKit.build_insert_vectors("docs_doc_vectors", vectors,
                                                      ids=row_ids, extra_data=metas,
                                                      id_col="rowid"):
        session.execute(text(sql), params)
# 正常退出统一提交；异常自动回滚（多处写入保持同一事务）

SQLite 的 DBAPI 变体（raw_connection() 支持 :name 命名参数）：

conn = SQLiteKit.raw_connection()          # 调用方负责 commit / close
try:
    for sql, params in SQLiteKit.build_insert_vectors("item_vectors", vectors,
                                                      ids=row_ids, extra_data=metas):
        conn.executemany(sql, params)
    conn.commit()
finally:
    conn.close()

⚠ 两点约束：① build_insert_vectors 输出为 :name 风格，PG / MySQL 的原生 DBAPI 不支持 （psycopg2 / pymysql 需 %(name)s），只能经 transaction() 的 session 执行； ② SQLite 上持有未提交写事务时再调 kit 的自提交方法（如 insert_vectors）会因另一个连接 拿不到写锁而失败（OperationalError: database is locked，约 6.4s 后才报错）—— 把写入并入同一个 transaction()（如上）即可避免。

相似度换算建议：SQLite 仅支持 L2，若向量入库前已归一化，则 cos = 1 − L2²/2； PG 可直接用 distance_type="COSINE"。

3. CRUDBase（配合 session_scope，不自动提交）

from kit.db import CRUDBase, session_scope

user_crud = CRUDBase(User)

with session_scope() as session:          # 正常退出提交、异常回滚
    user_crud.create(session, {"name": "张三"})
    user_crud.update(session, user_crud.get(session, 1), {"name": "李四"})

约定：SQLiteKit / PgKit / MySqlKit 方法自动提交；CRUDBase 方法不提交，事务边界由外层 session_scope 或 @transactional 决定。

4. 引擎 / 会话 / 事务 / 异常

from kit.db import transactional
from sqlmodel import Session

@transactional(retry_count=3)
def transfer(session: Session, from_id: int, to_id: int, amount: float) -> None:
    # 被装饰函数必须接受 session 参数；退出时自动提交，异常自动回滚并重试
    ...

5. 模型混入类与查询构建器

from sqlalchemy import BigInteger
from sqlmodel import Field, SQLModel

from kit.common_kit import CommonKit
from kit.db.models import TimestampMixin, SoftDeleteMixin, AuditMixin

class Article(TimestampMixin, SoftDeleteMixin, SQLModel, table=True):
    # 主键：64 位整数 + 雪花 ID（应用侧生成，不用数据库自增）
    id: int = Field(default_factory=CommonKit.generate_snowflake_id,
                    primary_key=True, sa_type=BigInteger)
    title: str

查询构建器需显式从子模块导入（未进入 kit.db 顶层导出）：

from kit.db.query import QueryBuilder, QueryCache, QueryOptimizer, cached_query

qb = QueryBuilder(User, session)
active = qb.filter(User.name == "张三").all()      # 链式不可变，每步返回新实例
page = qb.paginate(page=1, per_page=20)            # {"items", "total", ...}

kit.ai —— AI 多模型服务池

安装：pip install "sdpy-kit[ai]" 档案：项目根 config/ai_models.json（见文末「AI / MCP 档案文件」）

1. 同步门面 AI（推荐）

from kit.ai.runner import AI, shutdown

text = AI.chat_complete("用一句话介绍 Python")                 # 单轮对话
hits = AI.rerank("水声通信", ["文档A", "文档B"], top_n=2)      # 重排序
url = AI.generate_image("一只在写代码的猫")                    # 文生图
AI.use("chat", "glm")                                          # 切换当前档案（进程内长期生效）
print(AI.available("chat"))                                    # 查看各档案状态快照
shutdown()                                                     # 应用退出前释放资源（幂等）

from kit.ai.runner import AI

with AI.chat_stream([{"role": "user", "content": "写一首短诗"}]) as stream:
    for token in stream:      # 逐 token 消费
        print(token, end="", flush=True)

门面仅面向同步上下文；若调用点本身在 async 函数内，请直接 await 原生异步服务，避免阻塞事件循环线程。 模块级另有 kit.ai.runner.reload_config(...)、run_sync(coro_factory, *, timeout=120.0)、shutdown()。

2. 原生异步服务

from kit.ai import (
    ChatService, EmbeddingService, RerankService, ImageService,   # 服务类
    ChatRequest, EmbeddingRequest, RerankRequest, ImageRequest,   # 请求模型
    RerankHit, ModelBuilder,                                      # 结果类型 / 构建器
)

service = ModelBuilder.build_chat()                 # 老按类型构建：build_chat / build_embedding / build_rerank / build_image
answer = await service.ask(messages)                # 或 await service.complete(prompt)
await service.close()

3. 异常

from kit.ai.exceptions import (
    AIError, AIConfigError, AIRequestError, AITimeoutError,
    AIAggregateError, AIResponseError,
)

kit.mcp_kit —— 通用 MCP 客户端

安装：pip install "sdpy-kit[mcp]" 档案：项目根 config/mcp_servers.json；token 值放 .env（档案里只写环境变量名）

from kit.mcp_kit import McpKit

tools = McpKit.list_tools("aminer")                                  # [{name, description}]
result = McpKit.call_tool("search_paper_by_title", {"title": "水声通信", "size": 10})
print(result.data, result.elapsed_ms)

# 直连临时服务器（不经档案）
r = McpKit.call_tool_url("https://example.com/mcp", "query", {}, token="...")

# 异步环境 / 一次握手复用多次调用
async with McpKit.session("aminer") as call:
    r1 = await call("search_paper_by_title", {"title": "A"})
    r2 = await call("get_papers_info", {"id": "1"})

传输方式仅 streamable-http（默认）与 sse；档案配 fallback_url 时主端点传输层失败会自动回退（业务失败不回退）。

kit.redis_kit —— Redis 工具

安装：pip install "sdpy-kit[redis]"

from kit.redis_kit import RedisKit

RedisKit.set("user:1", "clark", expire=60)
RedisKit.get("user:1")                       # -> "clark"
RedisKit.zadd("rank", {"alice": 1.0, "bob": 2.0})
RedisKit.zrange("rank", 0, -1, withscores=True)
RedisKit.close()

RedisValue = str | int | float | bytes；连接池 decode_responses=True，内置指数退避重试。

kit.file_kit —— 文件工具

相对路径均相对于 FILE_STORAGE_ROOT（默认 ./data）。

from kit.file_kit import FileKit

FileKit.write_file("docs/a.txt", "hello")         # mode="overwrite"（默认）/ "append"
text = FileKit.read_file("docs/a.txt")
path = FileKit.get_full_path("docs/a.txt")        # -> Path 绝对路径
for root, dirs, files in FileKit.walk_dir("docs"):
    print(root, files)
FileKit.open_with_default_app(path)               # 系统默认程序打开：-> "opened" / "revealed"

kit.common_kit —— 通用工具

from kit.common_kit import CommonKit

CommonKit.configure(datacenter_id=1, machine_id=1)      # 各限 1–31
CommonKit.generate_snowflake_id()                       # 63 位雪花 ID（int）
CommonKit.generate_snow_id()                            # 雪花 ID（str）
CommonKit.generate_timestamp_id()                       # 20 位时间戳 ID
CommonKit.format_current_time("yyyy-MM-dd")

kit.anydoc_kit —— 文档转 Markdown

安装：pip install "sdpy-kit[anydoc]"（支持 .doc .docx .ppt .pptx .xls .xlsx .csv .pdf）

from kit.anydoc_kit import AnyDocKit

markdown = AnyDocKit.convert_to_md("report.docx")     # 直接转换，失败抛异常
result = AnyDocKit.convert("report.pdf")              # 安全转换，异常封装为结果对象
if result.success:
    print(result.markdown)
AnyDocKit.convert_to_file("report.docx", "out/report.md")
md = AnyDocKit.convert_bytes_to_md(file_bytes, "csv") # 字节内容直接转，无需落盘
md = AnyDocKit.convert_to_md("scan.pdf", ocr="hosted")  # 需 FIRECRAWL_API_KEY

ConvertResult(success, markdown, file_path, source_format, error)

kit.docx_kit —— Word 模板动态填充

安装：pip install "sdpy-kit[docx]"

from kit.docx_kit import DocxKit, DocxReport, DocxSection, DocxParagraph, DocxTable

DocxKit.create_template("templates/report.docx", title="{{ title }}")   # 生成模板骨架
DocxKit.unify_font("templates/report.docx", font_name="宋体")
DocxKit.unify_heading_color("templates/report.docx")

report = DocxReport(
    sections=(
        DocxSection(
            title="第一章 概述",
            level=1,
            blocks=(
                DocxParagraph("本章介绍项目背景。"),
                DocxTable(headers=("名称", "数值"), rows=(("甲", "1"),)),
            ),
        ),
    ),
    variables={"project": "SDPY"},
)
DocxKit.render("templates/report.docx", "out/report.docx", report)

数据结构：DocxReport(sections, variables)、DocxSection(title, level=1, blocks=())、DocxParagraph(text)、DocxListItem(text, level=1, ordered=False)、DocxTable(headers, rows=(), style=None)

kit.md2doc_kit —— Markdown 转 Word / PDF / HTML

安装：pip install "sdpy-kit[md2doc]"（PDF 需系统已安装 mainfont 指定字体，默认 SimSun）

from kit.md2doc_kit import Md2DocKit

Md2DocKit.convert_md_to_docx("report.md", "out.docx", generate_toc=True)
Md2DocKit.convert_md_to_html("report.md", title="项目报告")
Md2DocKit.convert_md_to_pdf("report.md", mainfont="KaiTi")

dest_path=None 时输出到源文件同目录；同一源文件不支持并发转换。

kit.mineru_kit —— MinerU 文档解析

安装：pip install "sdpy-kit[mineru]"（精准解析需 MINERU_API_KEY，Agent 轻量解析免登录）

from kit.mineru_kit import MinerUKit

# 精准解析：公网 URL / 本地文件
result = MinerUKit.parse_url("https://cdn-mineru.openxlab.org.cn/demo/example.pdf", model_version="vlm")
if result.state == "done":
    zip_path = MinerUKit.download_zip(result.full_zip_url, "./output")

results = MinerUKit.parse_files(["report1.pdf", "report2.docx"])

# Agent 轻量解析（免登录、单文件 ≤10MB、≤20 页）
r = MinerUKit.agent_parse_url(url, page_range="1-10")
if r.state == "done":
    md_path = MinerUKit.download_markdown(r.markdown_url, "./output")

可选参数（model_version / is_ocr / enable_formula / enable_table / language / page_ranges / extra_formats / no_cache 等）详见各方法 docstring。 结果类型：TaskResult、BatchTaskResult、BatchFile、BatchUploadInfo、AgentParseResult、AgentUploadInfo。

kit.marker_kit —— 本地文档解析（Markdown + Images）

安装：pip install "sdpy-kit[marker]"（依赖 marker-pdf[full]，含 torch 等 80+ 包，体积较大）

from kit.marker_kit import MarkerKit

# 落盘 md + images（推荐）：源文件同目录下建同名子目录
md_path = MarkerKit.convert_to_dir("report.pdf")              # -> report/report.md + 图片
md_path = MarkerKit.convert_to_dir("report.pdf", "out/doc")   # 指定产物目录

# 只取 Markdown 文本（图片不落盘）
md = MarkerKit.convert_to_md("report.pdf")

# 安全转换（不抛异常，返回结果对象）
result = MarkerKit.convert("report.pdf", "out/doc")
if result.success:
    print(result.md_path, len(result.image_paths), result.page_count)

# 跳过 VLM（无需 llama.cpp）
md = MarkerKit.convert_to_md("scan.pdf", disable_ocr=True)

MarkerKit.shutdown()   # 释放模型与推理服务（幂等）

产物：<output_dir>/<源文件名>.md + 图片文件（md 内相对路径引用），不生成 _meta.json。 结果类型：MarkerResult（success / markdown / file_path / output_dir / md_path / image_paths / page_count / error）；异常：MarkerError。 mode 缺省时不写入配置，交由 marker 按设备自动选择；首次使用会下载模型权重（默认走 MARKER_HF_ENDPOINT 镜像）。 非 NVIDIA 平台完整 VLM 能力（公式 / 扫描页 OCR）需 llama.cpp 的 llama-server，否则用 disable_ocr=True 跳过。

kit.zvec_kit —— 本地向量库

安装：pip install "sdpy-kit[zvec]"

from kit.zvec_kit import ZvecKit

schema = ZvecKit.make_text_embedding_schema("docs", dim=768)
ZvecKit.create_collection("docs", schema)
ZvecKit.insert("docs", {"id": "d1", "vectors": {"embedding": [0.1] * 768}, "fields": {"year": 2024}})

results = ZvecKit.vector_query("docs", field="embedding", vector=[0.3] * 768, topk=10)
ZvecKit.flush("docs")

文档字典键：id（必需）、vectors（可选）、fields（可选）；filter_expr 中相等用单等号 =。 异常：ZvecError（独立于 kit.db 的 DBError 体系）。

kit.wx_article_kit —— 微信公众号文章抓取存储

零第三方依赖（标准库 urllib + 内置 HTML→Markdown 转换器），无需 extra，按需 import：

from kit.wx_article_kit import WxArticleKit

# 抓取解析（不落盘）
result = WxArticleKit.fetch("https://mp.weixin.qq.com/s/xxxx")
if result.ok:
    print(result.info.title, result.info.author, result.info.publish_time)

# 抓取并存储：<output_dir>/<file_name>/ 下 <file_name>.md + images/
saved = WxArticleKit.save(url, output_dir="./output", file_name="12345")
print(saved.md_path, saved.images_saved)

# 需要结构化元信息时再产出 meta.json（默认不产出）
WxArticleKit.save(url, "./output", "12345", write_meta=True)

落盘结构：

<output_dir>/<file_name>/
├── <file_name>.md     # 正文 Markdown（YAML 头：title / author / publish_time / source_url）
├── images/            # 图片转存（NNN.ext，正文引用已本地化）
└── meta.json          # 可选：结构化元信息（write_meta=True 时产出）

• 元信息默认随 Markdown 的 YAML 头落盘（头部 publish_time 为中文日期，如 2026年8月4日）； ArticleInfo.publish_time 与 meta.json 内仍为 ISO 8601（结构化数据）。
• 无登录态抓取：命中微信风控（环境异常 / 访问过于频繁）时返回 ok=False + 原因，不破解验证码； 抓取或图片下载失败不抛异常，单张图片失败时正文保留原远程 URL。
• 同名子目录覆盖写（不追加序号）；file_name 带 .md 后缀会自动去除，为空则抛 ValueError。

AI / MCP 档案文件

AI 与 MCP 功能需要在项目根目录提供两个 JSON 档案文件：

• config/ai_models.json —— chat / embedding / rerank / image 四类型多厂商模型档案
• config/mcp_servers.json —— MCP 服务器档案

文件缺失时，首次使用相应功能会报错，报错信息中附带可直接复制的最小模板； 其中 api_key / token_env 字段填写环境变量名，真实密钥配置在 .env。

ai_models.json 可选字段

候选链切换语义（两类，勿混淆）

• 跨调用切换：AI.use() 与其它请求产生的失败冷却均只影响下一次请求——每次调用的候选计划在开始时定格；
• 一次调用内降级：本次调用按开始时定格的计划串行尝试（首帧前失败 / 首帧超时切下一候选），中途的 use / 冷却变化不影响它。

全部候选失败（或整链预算耗尽）抛 AIAggregateError（AIRequestError 子类）： 全候选状态码一致时透传（如整链 429 可用 exc.status_code == 429 判定）， 429 的 Retry-After 解析后挂于 exc.retry_after（冷却钳制 5–300s）。

mcp_servers.json 最小模板

{
  "default_server": "aminer",
  "servers": {
    "aminer": {
      "transport": "streamable-http",
      "url": "https://mcp.aminer.cn/mcp",
      "token_env": "AMINER_TOKEN",
      "timeout": 60
    }
  }
}

License

MIT