muse-agent-example/src/muse/接入/审校操作.py

"""作者审校命令合同；HTTP与CLI共用类型及具名业务转交。"""

from dataclasses import asdict, replace
from typing import Literal

from pydantic import BaseModel, ConfigDict, Field, StrictInt

from muse.共享.调用身份 import 内容用途
from muse.审校修订.接口 import (
    修订请求,
    声音定位,
    声音提案请求,
    案例复核请求,
    案例定位,
    案例采集请求,
    模型修订授权,
    规则状态命令,
)
from muse.正式变更.接口 import 作者动作
from muse.正文写作.接口 import 选区坐标, 选段写作授权


class 目标正文(BaseModel):
    model_config = ConfigDict(extra="forbid")
    work_id: str = Field(min_length=1)
    chapter_id: str = Field(min_length=1)
    revision: StrictInt = Field(ge=1)
    document_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    branch_id: str = Field(default="main", min_length=1)


class 审校发起请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    target: 目标正文
    config_id: str = Field(min_length=1)
    candidate_id: str | None = None
    shard_size: StrictInt = Field(default=4000, ge=2, le=12000)
    context_size: StrictInt = Field(default=240, ge=0, le=2000)


class 诊断请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    target: 目标正文
    preview_candidates: bool = False


class 声音保存请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    request: 声音提案请求


class 声音决定请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    target: 声音定位
    review_id: str
    review_hash: str
    action: Literal["adopt", "reject", "defer"]


def 执行审校发起(app, 身份, 请求: 审校发起请求):
    from muse.编排.完整审校 import 发起完整审校

    t = 请求.target
    return 发起完整审校(
        app,
        replace(身份, 内容用途=内容用途.检测),
        请求.command_id,
        work_id=t.work_id,
        chapter_id=t.chapter_id,
        revision=t.revision,
        document_hash=t.document_hash,
        分支=t.branch_id,
        配置ID=请求.config_id,
        candidate_id=请求.candidate_id,
        核心上限=请求.shard_size,
        上下文字数=请求.context_size,
    )


def 执行声音决定(app, 身份, 请求: 声音决定请求):
    return app.要求审校().决定声音(
        身份,
        请求.command_id,
        请求.target,
        审阅ID=请求.review_id,
        审阅哈希=请求.review_hash,
        动作=作者动作(请求.action),
    )


class 修订保存请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    request: 修订请求


class 模型修订请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    config_id: str = Field(min_length=1)
    request: 模型修订授权 | 选段写作授权


class 有限返修请求(模型修订请求):
    max_rounds: StrictInt = Field(ge=1, le=5)


class 返修选择请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    session_id: str = Field(min_length=1)
    expected_round: StrictInt = Field(ge=1, le=5)
    candidate_id: str = Field(min_length=1)
    choice: Literal["original", "candidate", "retry"]


def 执行有限返修(app, 身份, 请求: 有限返修请求):
    from muse.编排.审校修订 import 发起有限返修

    return 发起有限返修(
        app, 身份, 请求.command_id, 请求.request, 配置ID=请求.config_id, 最大轮数=请求.max_rounds
    )


def 执行返修选择(app, 身份, 请求: 返修选择请求):
    from muse.审校修订.接口 import 返修决定
    from muse.编排.审校修订 import 决定有限返修

    return 决定有限返修(
        app,
        身份,
        请求.command_id,
        请求.session_id,
        返修决定(请求.expected_round, 请求.candidate_id, 请求.choice),
    )


class 选段定位请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    target: 目标正文
    selections: tuple[选区坐标, ...]


def 执行选段定位(app, 身份, 请求: 选段定位请求):
    t = 请求.target
    result = app.要求正文().读取选段定位(
        身份, t.work_id, t.chapter_id, t.branch_id, t.revision, t.document_hash, 请求.selections
    )
    return [asdict(s) for s in result]


def 执行模型修订(app, 身份, 请求: 模型修订请求):
    from muse.编排.审校修订 import 发起模型修订

    return 发起模型修订(app, 身份, 请求.command_id, 请求.request, 配置ID=请求.config_id)


class 案例保存请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    request: 案例采集请求


class 案例决定请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    request: 案例复核请求
    review_id: str
    review_hash: str
    action: Literal["adopt", "reject", "defer"]


class 案例撤回请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    command_id: str = Field(min_length=1)
    target: 案例定位


def 执行案例决定(app, 身份, r: 案例决定请求):
    return app.要求审校().决定案例(
        身份,
        r.command_id,
        r.request,
        审阅ID=r.review_id,
        审阅哈希=r.review_hash,
        动作=作者动作(r.action),
    )


class 规则审阅请求(BaseModel):
    model_config = ConfigDict(extra="forbid")
    request: 规则状态命令


class 规则决定请求(规则审阅请求):
    command_id: str = Field(min_length=1)
    review_id: str = Field(min_length=1)
    review_hash: str = Field(pattern=r"^[0-9a-f]{64}$")
    action: Literal["adopt", "reject", "defer"]


def 执行规则决定(app, 身份, 请求: 规则决定请求):
    return app.要求审校().决定规则状态(
        身份,
        请求.command_id,
        请求.request,
        审阅ID=请求.review_id,
        审阅哈希=请求.review_hash,
        动作=作者动作(请求.action),
    )