Skip to content

ccran / lufa-contract

Go to a project
Commit 12354a37 by ccran
Options

feat: update review flow;

parent 8371d9cd
Showing with 30 additions and 532 deletions
data/rules.xlsx
No preview for this file type
skills/review-flow-skill/SKILL.md
--- ---
name: review-flow-skill name: review-flow-skill
description: 合同审查编排 Skill。用于把 URL 或本地合同文件下载/解析为 txt，逐条覆盖 rules.xlsx 中的审查规则，调用 review-llm-skill 完成路由、审查、反思、合并，并将结果文件上传。 description: 合同审查编排说明 Skill。仅提供合同审查的流程、规则覆盖、状态落盘和结果产出要求；LLM 需要根据这些编排规则自行选择并调用合适的 Skill 完成审查。
--- ---
# Review Flow Skill # Review Flow Skill
## 定位 ## 定位
`review-flow-skill` 是合同审查的流程编排层。它不直接实现 HTTP、文档解析、Excel 读写或 LLM 提示词，而是按固定顺序组合以下 Skill： `review-flow-skill` 是合同审查的编排说明层。它本身不执行下载、解析、Excel 读写、LLM 审查、上传，也不要求调用固定入口脚本。
使用本 Skill 时，LLM 必须阅读并遵循下方编排规则，自行根据当前任务、输入类型和可用能力选择合适的 Skill 或工具完成审查。常见协作 Skill 包括：
- `http-skill`：URL 输入时下载源文件；审查完成后上传结果文件。 - `http-skill`：URL 输入时下载源文件；审查完成后上传结果文件。
- `doc-excel-skill`：把 Word/PDF/Excel/CSV 解析成 UTF-8 txt；读取 `assets/rules.xlsx`；把 JSON 结果写入 Excel。 - `doc-excel-skill`：把 Word/PDF/Excel/CSV 解析成 UTF-8 txt；读取 `assets/审查规则.xlsx`；把 JSON 结果写入 Excel。
- `review-llm-skill`：执行规则路由、具体审查、反思复核、结果合并。 - `review-llm-skill`：执行规则路由、具体审查、反思复核、结果合并。
如果某个协作 Skill、脚本或工具不可用，LLM 应在不破坏编排约束的前提下选择等价能力完成相同步骤，并在结果中说明替代方式。
## 标准流程 ## 标准流程
1. 输入识别 1. 输入识别
- 如果输入是 `http://` 或 `https://` URL，必须先调用 `http-skill` 下载到本地工作目录。
- 如果输入是本地路径，直接进入解析步骤。
- 如果输入是 `http://` 或 `https://` URL，应先使用可用的 HTTP 下载能力下载到本地工作目录。
- 如果输入是本地路径，直接进入解析步骤。
2. 文档解析 2. 文档解析
- 优先把待审文件解析为 `.txt`，减少后续上下文占用。 - 优先把待审文件解析为 `.txt`，减少后续上下文占用。
- `.txt` 直接复用。 - `.txt` 直接复用。
- `.doc`、`.docx`、`.wps`、`.pdf` 使用 `doc-excel-skill/scripts/doc_tool.py` 转 txt。 - `.doc`、`.docx`、`.wps`、`.pdf` 应使用可用的文档解析能力转 txt。
- `.xlsx`、`.xls`、`.csv`、`.tsv` 使用 `doc-excel-skill/scripts/excel_tool.py` 读取各 sheet 后序列化为 txt。 - `.xlsx`、`.xls`、`.csv`、`.tsv` 应使用可用的表格解析能力读取各 sheet 后序列化为 txt。
3. 规则加载 3. 规则加载
- 默认从 `skills/review-flow-skill/assets/rules.xlsx` 读取审查规则。
- 默认从 `skills/review-flow-skill/assets/审查规则.xlsx` 读取审查规则。
- 未指定 sheet 时读取所有 sheet。 - 未指定 sheet 时读取所有 sheet。
- 必须覆盖到每一条非空规则。流程不得因为路由未命中而跳过“规则覆盖记录”。 - 必须覆盖到每一条非空规则。流程不得因为路由未命中而跳过“规则覆盖记录”。
4. 状态记录 4. 状态记录
- 每次运行在工作目录生成 `state.json`。 - 每次运行在工作目录生成 `state.json`。
- 状态至少记录：源文件、txt 文件、规则文件、规则总数、当前规则序号、已完成规则、输出文件路径。 - 状态至少记录：源文件、txt 文件、规则文件、规则总数、当前规则序号、已完成规则、输出文件路径。
- 每完成一条规则立即写回状态，用于断点续审。 - 每完成一条规则立即写回状态，用于断点续审。
5. 逐条规则审查 5. 逐条规则审查
- 按规则顺序逐条推进。 - 按规则顺序逐条推进。
- 对每条规则，先对 txt 分块调用 `review-llm-skill` 的 `router` 动作。 - 对每条规则，先对 txt 分块执行路由判断。
- 只有路由命中的分块才调用 `review` 动作。 - 只有路由命中的分块才执行具体审查。
- 即使所有分块均未命中，该规则也必须标记为已覆盖，并记录路由未命中。 - 即使所有分块均未命中，该规则也必须标记为已覆盖，并记录路由未命中。
6. 单规则反思 6. 单规则反思
- 每条规则的分块审查完成后，将该规则的原始 findings 放入 `context.findings`，调用 `reflect` 动作复核。
- 反思结果写入中间结果文件。
- 每条规则的分块审查完成后，将该规则的原始 findings 放入反思上下文，执行复核。
- 反思结果写入中间结果文件。
7. 全局合并 7. 全局合并
- 所有规则审查完毕后，对反思后的 findings 按 `rule_name`、`result`、`original_text` 聚合。 - 所有规则审查完毕后，对反思后的 findings 按 `rule_name`、`result`、`original_text` 聚合。
- 同组多条 findings 调用 `merge` 动作融合 `issue` 和 `suggestion`。 - 同组多条 findings 需要融合 `issue` 和 `suggestion`。
- 输出最终 JSON 和 Excel 汇总文件。 - 输出最终 JSON 和 Excel 汇总文件。
8. 结果上传 8. 结果上传
- 默认使用 `http-skill` 上传最终 Excel 文件。
- 如调用方只需要本地文件，可传 `--no-upload`。
## 工具文件 - 如调用方需要上传结果，应使用可用的上传能力上传最终 Excel 文件。
- 如调用方只需要本地文件，可仅保留本地输出。
## 编排资源
- `scripts/review_flow.py`：审查流程编排 CLI。 - `assets/审查规则.xlsx`：默认审查规则库。
- `assets/rules.xlsx`：默认审查规则库。
## 输出文件 ## 输出文件
运行目录默认位于 `skills/review-flow-skill/outputs/run-YYYYmmdd-HHMMSS/`，包含： 每次审查应创建独立运行目录，例如 `skills/review-flow-skill/outputs/{输入文件名}/`，包含：
- `input.txt`：解析后的合同文本。 - `input.txt`：解析后的合同文本。
- `state.json`：断点续审状态。 - `state.json`：断点续审状态。
...@@ -70,59 +73,12 @@ description: 合同审查编排 Skill。用于把 URL 或本地合同文件下 ...@@ -70,59 +73,12 @@ description: 合同审查编排 Skill。用于把 URL 或本地合同文件下
- `findings_reflected.json`：按规则反思后的结果。 - `findings_reflected.json`：按规则反思后的结果。
- `final_findings.json`：全局合并后的最终 JSON 结果。 - `final_findings.json`：全局合并后的最终 JSON 结果。
- `review_result.xlsx`：最终 Excel 汇总文件。 - `review_result.xlsx`：最终 Excel 汇总文件。
- `upload_result.json`：上传结果，仅在上传成功后生成。
## 使用示例
查看帮助：
```bash
python skills/review-flow-skill/scripts/review_flow.py --help
```
审查 URL 文件并上传结果：
```bash
python skills/review-flow-skill/scripts/review_flow.py \
https://example.com/contract.docx
```
审查本地文件，不上传：
```bash
python skills/review-flow-skill/scripts/review_flow.py \
demo/example.docx \
--no-upload
```
只审查指定 sheet：
```bash
python skills/review-flow-skill/scripts/review_flow.py \
demo/example.pdf \
--sheet-name 通用 \
--sheet-name 借款
```
从已有运行目录继续：
```bash
python skills/review-flow-skill/scripts/review_flow.py \
--resume skills/review-flow-skill/outputs/run-20260612-173000
```
调试时限制本次最多推进 3 条规则：
```bash
python skills/review-flow-skill/scripts/review_flow.py \
demo/example.docx \
--max-rules 3 \
--no-upload
```
## 约束 ## 约束
- 不在该 Skill 中重写下载、上传、文档解析、Excel 操作或 LLM 提示词能力。 - 不在该 Skill 中重写下载、上传、文档解析、Excel 操作或 LLM 提示词能力。
- 不把该 Skill 当作可直接执行的审查程序；它只定义编排规则。
- LLM 必须自行根据编排规则选择、组合并调用可用 Skill 或工具完成审查。
- 审查过程必须逐条规则推进并持久化状态。 - 审查过程必须逐条规则推进并持久化状态。
- 规则覆盖记录和最终结果必须落盘，避免只保存在上下文中。 - 规则覆盖记录和最终结果必须落盘，避免只保存在上下文中。
- 默认使用 txt 分块输入 LLM，避免一次性塞入完整合同造成上下文过大。 - 默认使用 txt 分块输入 LLM，避免一次性塞入完整合同造成上下文过大。
skills/review-flow-skill/scripts/review_flow.py deleted 100644 → 0
#!/usr/bin/env python3
"""Orchestrate URL/local-file contract review across every rule."""
from __future__ import annotations
import argparse
import json
import math
import shutil
import sys
from collections import defaultdict
from datetime import datetime
from pathlib import Path
from typing import Any
SCRIPT_DIR = Path(__file__).resolve().parent
SKILL_DIR = SCRIPT_DIR.parent
REPO_ROOT = SKILL_DIR.parents[1]
DEFAULT_RULES_FILE = SKILL_DIR / "assets" / "rules.xlsx"
DEFAULT_OUTPUT_ROOT = SKILL_DIR / "outputs"
HTTP_SCRIPT_DIR = REPO_ROOT / "skills" / "http-skill" / "scripts"
DOC_EXCEL_SCRIPT_DIR = REPO_ROOT / "skills" / "doc-excel-skill" / "scripts"
REVIEW_LLM_SCRIPT_DIR = REPO_ROOT / "skills" / "review-llm-skill" / "scripts"
for path in (HTTP_SCRIPT_DIR, DOC_EXCEL_SCRIPT_DIR, REVIEW_LLM_SCRIPT_DIR):
if str(path) not in sys.path:
sys.path.insert(0, str(path))
import excel_tool # type: ignore # noqa: E402
import http_util # type: ignore # noqa: E402
from doc_tool import doc_to_txt # type: ignore # noqa: E402
from segment_llm_action import run_segment_llm_action # type: ignore # noqa: E402
WORD_SUFFIXES = {".doc", ".docx", ".wps"}
PDF_SUFFIXES = {".pdf"}
TEXT_SUFFIXES = {".txt", ".md"}
TABLE_SUFFIXES = {".xlsx", ".xls", ".csv", ".tsv"}
TITLE_KEYS = ("审查项", "审查项名称", "规则名称", "规则标题", "title", "name")
def jdump(value: Any) -> str:
return json.dumps(value, ensure_ascii=False, indent=2)
def now_text() -> str:
return datetime.now().isoformat(timespec="seconds")
def is_url(value: str) -> bool:
return value.startswith(("http://", "https://"))
def write_json(path: Path, value: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(jdump(value) + "\n", encoding="utf-8")
def read_json(path: Path, default: Any) -> Any:
if not path.exists() or path.stat().st_size == 0:
return default
return json.loads(path.read_text(encoding="utf-8"))
def make_run_dir(output_root: Path) -> Path:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
run_dir = output_root / f"run-{stamp}"
index = 1
while run_dir.exists():
run_dir = output_root / f"run-{stamp}-{index}"
index += 1
run_dir.mkdir(parents=True)
return run_dir
def configure_http(args: argparse.Namespace) -> None:
http_util._configure_urls( # noqa: SLF001
http_util._strip(args.base_fastgpt_url), # noqa: SLF001
http_util._strip(args.base_backend_url), # noqa: SLF001
http_util._strip(args.outer_backend_url), # noqa: SLF001
)
http_util._configure_login(args.username, args.password) # noqa: SLF001
def resolve_source(source: str, run_dir: Path) -> Path:
if is_url(source):
downloaded = http_util.download_file(source, run_dir / "download")
if not downloaded:
raise RuntimeError(f"download failed: {source}")
return Path(downloaded)
path = Path(source).expanduser()
if not path.is_absolute():
path = Path.cwd() / path
if not path.exists():
raise FileNotFoundError(path)
return path
def excel_to_txt(path: Path, output: Path) -> Path:
chunks: list[str] = []
sheets = excel_tool.list_sheets(str(path)) if path.suffix.lower() in {".xlsx", ".xls"} else [None]
for sheet in sheets:
rows = excel_tool.load_excel(str(path), sheet=sheet, header=True)
title = sheet or path.name
chunks.append(f"# Sheet: {title}\n{jdump(rows)}")
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text("\n\n".join(chunks), encoding="utf-8")
return output
def parse_to_txt(source_file: Path, run_dir: Path) -> Path:
suffix = source_file.suffix.lower()
output = run_dir / "input.txt"
if suffix in TEXT_SUFFIXES:
if source_file.resolve() != output.resolve():
shutil.copyfile(source_file, output)
return output
if suffix in WORD_SUFFIXES or suffix in PDF_SUFFIXES:
return Path(doc_to_txt(str(source_file), str(output)))
if suffix in TABLE_SUFFIXES:
return excel_to_txt(source_file, output)
raise ValueError(f"unsupported input file type: {suffix}")
def compact_rule(row: dict[str, Any], sheet_name: str, row_index: int) -> dict[str, Any]:
cleaned = {str(k).strip(): v for k, v in row.items() if str(k).strip() and v not in (None, "")}
title = next((str(cleaned[key]).strip() for key in TITLE_KEYS if cleaned.get(key)), "")
if not title:
title = next((str(value).strip() for value in cleaned.values() if value not in (None, "")), "")
cleaned["_sheet_name"] = sheet_name
cleaned["_row_index"] = row_index
cleaned["_rule_id"] = f"{sheet_name}:{row_index}:{title or 'untitled'}"
cleaned["_title"] = title or cleaned["_rule_id"]
return cleaned
def load_rules(rules_file: Path, sheet_names: list[str] | None) -> list[dict[str, Any]]:
selected_sheets = sheet_names or excel_tool.list_sheets(str(rules_file))
rules: list[dict[str, Any]] = []
for sheet_name in selected_sheets:
rows = excel_tool.search_rows(str(rules_file), sheet_name, {})
for index, row in enumerate(rows, start=2):
if not isinstance(row, dict):
continue
rule = compact_rule(row, sheet_name, index)
if any(not str(key).startswith("_") for key in rule):
rules.append(rule)
return rules
def split_text(text: str, chunk_size: int) -> list[dict[str, Any]]:
if chunk_size <= 0:
raise ValueError("--chunk-size must be > 0")
total = max(1, math.ceil(len(text) / chunk_size))
return [
{
"chunk_index": index,
"chunk_total": total,
"start": index * chunk_size,
"end": min((index + 1) * chunk_size, len(text)),
"text": text[index * chunk_size : (index + 1) * chunk_size],
}
for index in range(total)
]
def selected_by_router(router_result: Any, rule: dict[str, Any]) -> bool:
if isinstance(router_result, list):
return bool(router_result)
if not isinstance(router_result, dict):
return False
selected = router_result.get("selected_items") or router_result.get("selected_rules") or []
if not isinstance(selected, list):
return False
if not selected:
return False
title = str(rule.get("_title", ""))
rule_id = str(rule.get("_rule_id", ""))
for item in selected:
if not isinstance(item, dict):
return True
text = jdump(item)
if title and title in text:
return True
if rule_id and rule_id in text:
return True
return True
def normalize_findings(value: Any) -> list[dict[str, Any]]:
if isinstance(value, dict):
if isinstance(value.get("findings"), list):
value = value["findings"]
else:
value = [value]
if not isinstance(value, list):
return []
return [item for item in value if isinstance(item, dict)]
def call_llm(action: str, rule: dict[str, Any], text: str) -> Any:
return run_segment_llm_action(action=action, rule=rule, text=text)
def review_rule(rule: dict[str, Any], chunks: list[dict[str, Any]]) -> dict[str, Any]:
routed_chunks: list[int] = []
router_results: list[dict[str, Any]] = []
raw_findings: list[dict[str, Any]] = []
router_rule = {
"title": rule.get("_title"),
"rule_id": rule.get("_rule_id"),
"candidate_rules": [rule],
}
for chunk in chunks:
router_result = call_llm("router", router_rule, chunk["text"])
hit = selected_by_router(router_result, rule)
router_results.append(
{
"chunk_index": chunk["chunk_index"],
"selected": hit,
"router_result": router_result,
}
)
if not hit:
continue
routed_chunks.append(chunk["chunk_index"])
review_rule_payload = {
**rule,
"context": {
"chunk_index": chunk["chunk_index"],
"chunk_total": chunk["chunk_total"],
},
}
review_result = call_llm("review", review_rule_payload, chunk["text"])
for finding in normalize_findings(review_result):
finding.setdefault("rule_name", rule.get("_title"))
finding["_rule_id"] = rule.get("_rule_id")
finding["_sheet_name"] = rule.get("_sheet_name")
finding["_row_index"] = rule.get("_row_index")
finding["_chunk_index"] = chunk["chunk_index"]
raw_findings.append(finding)
reflected_findings: list[dict[str, Any]] = []
if raw_findings:
reflect_payload = {
**rule,
"context": {
"findings": raw_findings,
"routed_chunks": routed_chunks,
},
}
evidence_text = "\n\n".join(
str(item.get("original_text", "")).strip()
for item in raw_findings
if item.get("original_text")
)
reflect_result = call_llm("reflect", reflect_payload, evidence_text)
reflected_findings = normalize_findings(reflect_result)
for finding in reflected_findings:
finding.setdefault("rule_name", rule.get("_title"))
finding["_rule_id"] = rule.get("_rule_id")
finding["_sheet_name"] = rule.get("_sheet_name")
finding["_row_index"] = rule.get("_row_index")
return {
"rule": rule,
"router_results": router_results,
"routed_chunks": routed_chunks,
"raw_findings": raw_findings,
"reflected_findings": reflected_findings,
}
def merge_findings(findings: list[dict[str, Any]]) -> list[dict[str, Any]]:
grouped: dict[tuple[str, str, str], list[dict[str, Any]]] = defaultdict(list)
for finding in findings:
key = (
str(finding.get("rule_name", "")),
str(finding.get("result", "")),
str(finding.get("original_text", "")),
)
grouped[key].append(finding)
merged: list[dict[str, Any]] = []
for (rule_name, result, original_text), items in grouped.items():
if len(items) == 1:
merged.append(items[0])
continue
merge_payload = {
"rule_name": rule_name,
"result": result,
"original_text": original_text,
"findings": [
{
"issue": item.get("issue", ""),
"suggestion": item.get("suggestion", ""),
"original_text": item.get("original_text", ""),
}
for item in items
],
}
merged_result = call_llm("merge", merge_payload, "")
base = dict(items[0])
if isinstance(merged_result, dict):
base["issue"] = merged_result.get("issue", base.get("issue", ""))
base["suggestion"] = merged_result.get("suggestion", base.get("suggestion", ""))
merged.append(base)
return merged
def save_outputs(run_dir: Path, progress: list[dict[str, Any]], raw: list[dict[str, Any]], reflected: list[dict[str, Any]], final: list[dict[str, Any]] | None = None) -> None:
write_json(run_dir / "rule_progress.json", progress)
write_json(run_dir / "findings_raw.json", raw)
write_json(run_dir / "findings_reflected.json", reflected)
if final is not None:
write_json(run_dir / "final_findings.json", final)
excel_tool.json_to_sheet(final, str(run_dir / "review_result.xlsx"), "审查结果")
def update_state(run_dir: Path, state: dict[str, Any]) -> None:
state["updated_at"] = now_text()
write_json(run_dir / "state.json", state)
def initial_state(args: argparse.Namespace, run_dir: Path, source_file: Path, text_file: Path, rules: list[dict[str, Any]]) -> dict[str, Any]:
return {
"created_at": now_text(),
"updated_at": now_text(),
"source": args.source,
"source_file": str(source_file),
"text_file": str(text_file),
"rules_file": str(Path(args.rules_file).resolve()),
"sheet_names": args.sheet_name or "ALL",
"chunk_size": args.chunk_size,
"current_rule_index": 0,
"total_rules": len(rules),
"completed_rules": [],
"outputs": {
"progress": str(run_dir / "rule_progress.json"),
"raw": str(run_dir / "findings_raw.json"),
"reflected": str(run_dir / "findings_reflected.json"),
"final": str(run_dir / "final_findings.json"),
"excel": str(run_dir / "review_result.xlsx"),
},
}
def run_new(args: argparse.Namespace) -> tuple[Path, dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
run_dir = make_run_dir(Path(args.output_root))
configure_http(args)
source_file = resolve_source(args.source, run_dir)
text_file = parse_to_txt(source_file, run_dir)
rules = load_rules(Path(args.rules_file), args.sheet_name)
state = initial_state(args, run_dir, source_file, text_file, rules)
write_json(run_dir / "rules.loaded.json", rules)
update_state(run_dir, state)
return run_dir, state, rules, split_text(text_file.read_text(encoding="utf-8"), args.chunk_size)
def run_resume(args: argparse.Namespace) -> tuple[Path, dict[str, Any], list[dict[str, Any]], list[dict[str, Any]]]:
run_dir = Path(args.resume).expanduser()
if not run_dir.is_absolute():
run_dir = Path.cwd() / run_dir
state = read_json(run_dir / "state.json", {})
if not state:
raise FileNotFoundError(run_dir / "state.json")
rules = read_json(run_dir / "rules.loaded.json", [])
if not rules:
rules = load_rules(Path(state["rules_file"]), None if state.get("sheet_names") == "ALL" else state.get("sheet_names"))
text = Path(state["text_file"]).read_text(encoding="utf-8")
return run_dir, state, rules, split_text(text, int(state["chunk_size"]))
def upload_result(args: argparse.Namespace, run_dir: Path) -> None:
if args.no_upload:
return
configure_http(args)
result = http_util.upload_file(run_dir / "review_result.xlsx")
write_json(run_dir / "upload_result.json", {"uploaded": result})
def run_review(args: argparse.Namespace) -> Path:
run_dir, state, rules, chunks = run_resume(args) if args.resume else run_new(args)
progress = read_json(run_dir / "rule_progress.json", [])
raw_findings = read_json(run_dir / "findings_raw.json", [])
reflected_findings = read_json(run_dir / "findings_reflected.json", [])
start = int(state.get("current_rule_index", 0))
stop = len(rules)
if args.max_rules is not None:
stop = min(stop, start + args.max_rules)
for rule_index in range(start, stop):
rule_result = review_rule(rules[rule_index], chunks)
progress.append(
{
"rule_index": rule_index,
"rule_id": rules[rule_index].get("_rule_id"),
"title": rules[rule_index].get("_title"),
"routed_chunks": rule_result["routed_chunks"],
"covered": True,
"raw_count": len(rule_result["raw_findings"]),
"reflected_count": len(rule_result["reflected_findings"]),
"completed_at": now_text(),
"router_results": rule_result["router_results"],
}
)
raw_findings.extend(rule_result["raw_findings"])
reflected_findings.extend(rule_result["reflected_findings"])
state["current_rule_index"] = rule_index + 1
state["completed_rules"].append(rules[rule_index].get("_rule_id"))
save_outputs(run_dir, progress, raw_findings, reflected_findings)
update_state(run_dir, state)
if int(state.get("current_rule_index", 0)) >= len(rules):
final_findings = merge_findings(reflected_findings)
save_outputs(run_dir, progress, raw_findings, reflected_findings, final_findings)
state["finalized_at"] = now_text()
update_state(run_dir, state)
upload_result(args, run_dir)
return run_dir
def parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="Run contract review flow over every rule.")
p.add_argument("source", nargs="?", help="URL or local contract file. Required unless --resume is used.")
p.add_argument("--resume", help="Resume from an existing run directory.")
p.add_argument("--rules-file", default=str(DEFAULT_RULES_FILE))
p.add_argument("--sheet-name", action="append", help="Rule sheet to load. May be repeated. Defaults to all sheets.")
p.add_argument("--output-root", default=str(DEFAULT_OUTPUT_ROOT))
p.add_argument("--chunk-size", type=int, default=6000)
p.add_argument("--max-rules", type=int, help="Only advance this many rules in the current invocation.")
p.add_argument("--no-upload", action="store_true")
p.add_argument("--base-fastgpt-url", default=http_util.DEFAULT_BASE_FASTGPT_URL)
p.add_argument("--base-backend-url", default=http_util.DEFAULT_BASE_BACKEND_URL)
p.add_argument("--outer-backend-url", default=http_util.DEFAULT_OUTER_BACKEND_URL)
p.add_argument("--username", default=http_util.DEFAULT_BACKEND_ADMIN_USERNAME)
p.add_argument("--password", default=http_util.DEFAULT_BACKEND_ADMIN_PASSWORD)
return p
def main(argv: list[str] | None = None) -> int:
args = parser().parse_args(argv)
if not args.resume and not args.source:
raise SystemExit("source is required unless --resume is used")
run_dir = run_review(args)
print(run_dir)
return 0
if __name__ == "__main__":
raise SystemExit(main())
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment