1222 lines
55 KiB
Python
1222 lines
55 KiB
Python
#!/usr/bin/env python3
|
|
"""Read and review an ACK-ready Feishu Base view through the official lark-cli.
|
|
|
|
The only mutation is a bounded draft write to configured Coordinator fields.
|
|
The adapter never reads the active profile and emits one JSON document only
|
|
on success.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import math
|
|
import os
|
|
import pwd
|
|
import re
|
|
import resource
|
|
import signal
|
|
import stat
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from approval_payload import approval_payload_hash, review_items
|
|
from validate_tasks import validate_builtin as validate_task_board
|
|
from yaml_subset import DuplicateKeyError, YamlSubsetError, load_json_unique, load_yaml_subset, make_unique_pyyaml_loader
|
|
|
|
LEGACY_REQUIRED_FIELDS = ("title", "actual", "expected", "stepsToReproduce", "acceptance", "attachments", "updatedAt")
|
|
LEGACY_OPTIONAL_FIELDS = ("priority", "fixLogic")
|
|
LEGACY_FIELD_ORDER = ("title", "actual", "expected", "stepsToReproduce", "acceptance", "priority", "attachments", "updatedAt", "fixLogic")
|
|
CLARIFIED_REQUIRED_FIELDS = ("title", "details", "problemStatement", "expectedOutcome", "acceptance", "intakeStatus", "ackTaskId", "attachments", "updatedAt")
|
|
CLARIFIED_FIELD_ORDER = CLARIFIED_REQUIRED_FIELDS
|
|
SOURCE_FACT_FIELDS = ("title", "actual", "expected", "updatedAt")
|
|
COORDINATOR_FIELDS = ("steps", "acceptance", "priority")
|
|
BUG_CONTENT_FIELDS = ("title", "actual", "expected", "fixLogic", *COORDINATOR_FIELDS)
|
|
MAX_PAGES = 100
|
|
MAX_RECORDS = 10_000
|
|
PAGE_SIZE = 100
|
|
MAX_CLI_STDOUT = 1024 * 1024
|
|
MAX_CLI_STDERR = 64 * 1024
|
|
MAX_ATTACHMENTS_PER_RECORD = 10
|
|
MAX_TOTAL_ATTACHMENTS = 100
|
|
MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
|
|
MAX_TOTAL_ATTACHMENT_BYTES = 200 * 1024 * 1024
|
|
MAX_ATTACHMENT_BATCH_SECONDS = 300
|
|
MAX_DRAFT_BYTES = 64 * 1024
|
|
SAFE_VALUE = re.compile(r"^[^\s\x00-\x1f]{1,256}$")
|
|
PROFILE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
|
|
RECORD_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,255}$")
|
|
SOURCE_REF = re.compile(r"^feishu-base:sha256:[0-9a-f]{64}$")
|
|
DRAFT_REVISION = re.compile(r"^sha256:[0-9a-f]{64}$")
|
|
WORKFLOWS = {"read-only-v1", "reviewed-writeback-v1", "clarified-writeback-v1"}
|
|
INTAKE_STATUSES = ("待整理", "需补充", "待审核", "已确认", "已导入")
|
|
TARGET_BASE_FIELDS = (
|
|
("标题", "text"), ("详细描述", "text"), ("附件", "attachment"),
|
|
("问题说明", "text"), ("期望效果", "text"), ("验收标准", "text"),
|
|
("处理状态", "select"), ("ACK任务ID", "text"), ("更新时间", "updated_at"),
|
|
)
|
|
|
|
|
|
class IntakeError(Exception):
|
|
pass
|
|
|
|
|
|
def account_home() -> Path:
|
|
"""Return the actual account home, never a caller-controlled HOME value."""
|
|
try:
|
|
home = Path(pwd.getpwuid(os.getuid()).pw_dir).resolve(strict=True)
|
|
except (KeyError, OSError) as exc:
|
|
raise IntakeError("cannot resolve current account home") from exc
|
|
if not home.is_dir():
|
|
raise IntakeError("current account home is not a directory")
|
|
return home
|
|
|
|
|
|
def trusted_lark_cli_dirs() -> list[Path]:
|
|
"""Fixed account and system locations; intentionally never consult PATH."""
|
|
home = account_home()
|
|
candidates = (
|
|
home / ".local" / "bin",
|
|
home / ".local" / "share" / "mise" / "shims",
|
|
home / ".cargo" / "bin",
|
|
Path("/home/linuxbrew/.linuxbrew/bin"),
|
|
Path("/usr/local/go/bin"),
|
|
Path("/usr/local/bin"),
|
|
Path("/usr/bin"),
|
|
Path("/bin"),
|
|
)
|
|
result: list[Path] = []
|
|
for candidate in candidates:
|
|
try:
|
|
resolved = candidate.resolve(strict=True)
|
|
except OSError:
|
|
continue
|
|
if resolved.is_dir() and resolved not in result:
|
|
result.append(resolved)
|
|
return result
|
|
|
|
|
|
def resolve_lark_cli() -> Path:
|
|
"""Resolve a safe lark-cli from the fixed trusted locations only."""
|
|
for directory in trusted_lark_cli_dirs():
|
|
candidate = directory / "lark-cli"
|
|
try:
|
|
candidate_metadata = os.lstat(candidate)
|
|
resolved = candidate.resolve(strict=True)
|
|
metadata = resolved.stat()
|
|
except OSError:
|
|
continue
|
|
if not (stat.S_ISREG(candidate_metadata.st_mode) or stat.S_ISLNK(candidate_metadata.st_mode)):
|
|
continue
|
|
if candidate_metadata.st_uid not in {0, os.getuid()}:
|
|
continue
|
|
if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK):
|
|
continue
|
|
if metadata.st_uid not in {0, os.getuid()} or stat.S_IMODE(metadata.st_mode) & 0o022:
|
|
continue
|
|
if resolved.name == "lark-cli":
|
|
return resolved
|
|
if not stat.S_ISLNK(candidate_metadata.st_mode):
|
|
continue
|
|
official_binary = official_npm_binary(resolved)
|
|
if official_binary is not None:
|
|
return official_binary
|
|
raise IntakeError("lark-cli is not installed in a trusted account or system directory")
|
|
|
|
|
|
def official_npm_binary(path: Path) -> Path | None:
|
|
"""Resolve a validated npm wrapper to its downloaded native CLI binary."""
|
|
if path.name != "run.js" or path.parent.name != "scripts":
|
|
return None
|
|
manifest = path.parent.parent / "package.json"
|
|
try:
|
|
metadata = manifest.stat()
|
|
if not stat.S_ISREG(metadata.st_mode):
|
|
return None
|
|
if metadata.st_uid not in {0, os.getuid()} or stat.S_IMODE(metadata.st_mode) & 0o022:
|
|
return None
|
|
package = json.loads(manifest.read_text(encoding="utf-8"))
|
|
except (OSError, UnicodeError, json.JSONDecodeError):
|
|
return None
|
|
if not (
|
|
isinstance(package, dict)
|
|
and package.get("name") == "@larksuite/cli"
|
|
and isinstance(package.get("bin"), dict)
|
|
and package["bin"].get("lark-cli") == "scripts/run.js"
|
|
):
|
|
return None
|
|
native = path.parent.parent / "bin" / "lark-cli"
|
|
try:
|
|
native_lstat = os.lstat(native)
|
|
resolved = native.resolve(strict=True)
|
|
metadata = resolved.stat()
|
|
except OSError:
|
|
return None
|
|
if not stat.S_ISREG(native_lstat.st_mode) or resolved.name != "lark-cli":
|
|
return None
|
|
if not stat.S_ISREG(metadata.st_mode) or not os.access(resolved, os.X_OK):
|
|
return None
|
|
if metadata.st_uid not in {0, os.getuid()} or stat.S_IMODE(metadata.st_mode) & 0o022:
|
|
return None
|
|
return resolved
|
|
|
|
|
|
def cli_environment() -> dict[str, str]:
|
|
"""Build a minimal environment so env credentials cannot override `--profile`."""
|
|
environment = {
|
|
"HOME": str(account_home()),
|
|
"PATH": os.pathsep.join(str(path) for path in trusted_lark_cli_dirs()),
|
|
}
|
|
for name in ("LANG", "LC_ALL", "LC_CTYPE"):
|
|
value = os.environ.get(name)
|
|
if value and "\x00" not in value and len(value) <= 256:
|
|
environment[name] = value
|
|
return environment
|
|
|
|
|
|
def load_board(path: Path) -> dict[str, Any]:
|
|
try:
|
|
text = path.read_text(encoding="utf-8")
|
|
if path.suffix.lower() == ".json":
|
|
value = load_json_unique(text)
|
|
else:
|
|
try:
|
|
import yaml # type: ignore
|
|
value = yaml.load(text, Loader=make_unique_pyyaml_loader(yaml))
|
|
except ImportError:
|
|
value = load_yaml_subset(text)
|
|
except (OSError, json.JSONDecodeError, DuplicateKeyError, YamlSubsetError) as exc:
|
|
raise IntakeError(f"cannot read task board: {exc}") from exc
|
|
except Exception as exc: # PyYAML errors are intentionally not exposed verbatim.
|
|
raise IntakeError("cannot parse task board") from exc
|
|
if not isinstance(value, dict):
|
|
raise IntakeError("task board must be an object")
|
|
return value
|
|
|
|
|
|
def load_draft(path: Path, workflow: str = "reviewed-writeback-v1") -> dict[str, Any]:
|
|
"""Load one bounded, regular JSON file with the two writable draft fields."""
|
|
descriptor: int | None = None
|
|
try:
|
|
before = path.lstat()
|
|
if not stat.S_ISREG(before.st_mode) or path.is_symlink():
|
|
raise IntakeError("draft input must be a regular file")
|
|
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
|
|
descriptor = os.open(path, flags)
|
|
metadata = os.fstat(descriptor)
|
|
if (
|
|
not stat.S_ISREG(metadata.st_mode)
|
|
or (before.st_dev, before.st_ino) != (metadata.st_dev, metadata.st_ino)
|
|
):
|
|
raise IntakeError("draft input changed while opening")
|
|
if metadata.st_size <= 0 or metadata.st_size > MAX_DRAFT_BYTES:
|
|
raise IntakeError("draft input size is invalid")
|
|
chunks: list[bytes] = []
|
|
total = 0
|
|
while total <= MAX_DRAFT_BYTES:
|
|
chunk = os.read(descriptor, min(64 * 1024, MAX_DRAFT_BYTES + 1 - total))
|
|
if not chunk:
|
|
break
|
|
chunks.append(chunk)
|
|
total += len(chunk)
|
|
content = b"".join(chunks)
|
|
if len(content) != metadata.st_size:
|
|
raise IntakeError("draft input changed while reading")
|
|
value = load_json_unique(content.decode("utf-8"))
|
|
except IntakeError:
|
|
raise
|
|
except (OSError, UnicodeError, json.JSONDecodeError, DuplicateKeyError) as exc:
|
|
raise IntakeError("cannot read draft input") from exc
|
|
finally:
|
|
if descriptor is not None:
|
|
os.close(descriptor)
|
|
if workflow == "clarified-writeback-v1":
|
|
required = {"problemStatement", "expectedOutcome", "acceptance"}
|
|
if not isinstance(value, dict) or set(value) != required:
|
|
raise IntakeError("draft input must contain exactly problemStatement, expectedOutcome and acceptance")
|
|
for field in ("problemStatement", "expectedOutcome"):
|
|
if not isinstance(value[field], str) or not value[field].strip():
|
|
raise IntakeError(f"draft {field} must be a non-empty string")
|
|
acceptance = value["acceptance"]
|
|
if not isinstance(acceptance, list) or not acceptance or any(
|
|
not isinstance(item, str) or not item.strip() for item in acceptance
|
|
):
|
|
raise IntakeError("draft acceptance must be a non-empty string list")
|
|
return {
|
|
"problemStatement": value["problemStatement"].strip(),
|
|
"expectedOutcome": value["expectedOutcome"].strip(),
|
|
"acceptance": [item.strip() for item in acceptance],
|
|
}
|
|
if not isinstance(value, dict) or set(value) != {"fixLogic", "acceptance"}:
|
|
raise IntakeError("draft input must contain exactly fixLogic and acceptance")
|
|
fix_logic = value["fixLogic"]
|
|
acceptance = value["acceptance"]
|
|
if not isinstance(fix_logic, str) or not fix_logic.strip():
|
|
raise IntakeError("draft fixLogic must be a non-empty string")
|
|
if (
|
|
not isinstance(acceptance, list)
|
|
or not acceptance
|
|
or any(not isinstance(item, str) or not item.strip() for item in acceptance)
|
|
):
|
|
raise IntakeError("draft acceptance must be a non-empty string list")
|
|
return {
|
|
"fixLogic": fix_logic.strip(),
|
|
"acceptance": [item.strip() for item in acceptance],
|
|
}
|
|
|
|
|
|
def config_from_board(board: dict[str, Any]) -> dict[str, Any]:
|
|
project = board.get("project")
|
|
if not isinstance(project, dict) or "bugIntake" not in project:
|
|
raise IntakeError("project.bugIntake is not configured")
|
|
config = project["bugIntake"]
|
|
if not isinstance(config, dict):
|
|
raise IntakeError("project.bugIntake must be an object")
|
|
allowed = {"provider", "workflow", "profile", "baseToken", "tableId", "viewId", "fields"}
|
|
unknown = sorted(set(config) - allowed)
|
|
if unknown:
|
|
raise IntakeError("bugIntake has unknown fields")
|
|
if config.get("provider") != "feishu-base":
|
|
raise IntakeError("bugIntake.provider must be feishu-base")
|
|
workflow = config.get("workflow", "read-only-v1")
|
|
if workflow not in WORKFLOWS:
|
|
raise IntakeError("bugIntake.workflow is invalid")
|
|
profile = config.get("profile")
|
|
if not isinstance(profile, str) or not PROFILE.fullmatch(profile):
|
|
raise IntakeError("bugIntake.profile is invalid")
|
|
for key in ("baseToken", "tableId", "viewId"):
|
|
value = config.get(key)
|
|
if not isinstance(value, str) or not SAFE_VALUE.fullmatch(value):
|
|
raise IntakeError(f"bugIntake.{key} is invalid")
|
|
fields = config.get("fields")
|
|
if workflow == "clarified-writeback-v1":
|
|
required = CLARIFIED_REQUIRED_FIELDS
|
|
supported = set(required)
|
|
else:
|
|
required = LEGACY_REQUIRED_FIELDS
|
|
supported = set(required) | set(LEGACY_OPTIONAL_FIELDS)
|
|
if (
|
|
not isinstance(fields, dict)
|
|
or not set(required).issubset(fields)
|
|
or not set(fields).issubset(supported)
|
|
):
|
|
raise IntakeError("bugIntake.fields must map all required and only supported logical fields")
|
|
if any(not isinstance(value, str) or not SAFE_VALUE.fullmatch(value) for value in fields.values()):
|
|
raise IntakeError("bugIntake.fields values are invalid")
|
|
if len(set(fields.values())) != len(fields):
|
|
raise IntakeError("bugIntake.fields values must be unique")
|
|
if workflow == "reviewed-writeback-v1":
|
|
missing_review_fields = {"fixLogic", "priority"} - set(fields)
|
|
if missing_review_fields:
|
|
raise IntakeError(
|
|
"reviewed writeback requires bugIntake.fields.fixLogic and priority"
|
|
)
|
|
return config
|
|
|
|
|
|
def limit_child_file_size(limit: int) -> None:
|
|
"""Bound regular-file writes by the CLI and any child spawned by its wrapper."""
|
|
_, hard = resource.getrlimit(resource.RLIMIT_FSIZE)
|
|
bounded = limit if hard == resource.RLIM_INFINITY else min(limit, hard)
|
|
resource.setrlimit(resource.RLIMIT_FSIZE, (bounded, bounded))
|
|
|
|
|
|
def run_cli(
|
|
args: list[str], *, allow_profile_list: bool = False, max_file_bytes: int = 0,
|
|
timeout: float = 60, cwd: Path | None = None,
|
|
) -> Any:
|
|
"""Run the official CLI and accept only explicit successful JSON shapes."""
|
|
executable = resolve_lark_cli()
|
|
file_limit = max(MAX_CLI_STDOUT, MAX_CLI_STDERR, max_file_bytes)
|
|
with tempfile.TemporaryFile() as stdout_file, tempfile.TemporaryFile() as stderr_file:
|
|
try:
|
|
process = subprocess.Popen(
|
|
[str(executable), *args],
|
|
shell=False,
|
|
stdin=subprocess.DEVNULL,
|
|
stdout=stdout_file,
|
|
stderr=stderr_file,
|
|
env=cli_environment(),
|
|
cwd=cwd,
|
|
start_new_session=True,
|
|
preexec_fn=lambda: limit_child_file_size(file_limit),
|
|
)
|
|
try:
|
|
returncode = process.wait(timeout=timeout)
|
|
except subprocess.TimeoutExpired as exc:
|
|
try:
|
|
os.killpg(process.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
process.wait()
|
|
raise IntakeError("lark-cli failed to execute") from exc
|
|
except (OSError, subprocess.SubprocessError) as exc:
|
|
raise IntakeError("lark-cli failed to execute") from exc
|
|
stdout_size = os.fstat(stdout_file.fileno()).st_size
|
|
stderr_size = os.fstat(stderr_file.fileno()).st_size
|
|
if stdout_size > MAX_CLI_STDOUT or stderr_size > MAX_CLI_STDERR:
|
|
raise IntakeError("lark-cli output exceeded the safety limit")
|
|
if returncode:
|
|
raise IntakeError("lark-cli command failed")
|
|
stdout_file.seek(0)
|
|
try:
|
|
stdout = stdout_file.read(MAX_CLI_STDOUT + 1).decode("utf-8")
|
|
except UnicodeDecodeError as exc:
|
|
raise IntakeError("lark-cli returned malformed JSON") from exc
|
|
try:
|
|
value = json.loads(
|
|
stdout,
|
|
parse_constant=lambda value: (_ for _ in ()).throw(
|
|
ValueError(f"non-finite JSON constant: {value}")
|
|
),
|
|
)
|
|
except (json.JSONDecodeError, ValueError) as exc:
|
|
raise IntakeError("lark-cli returned malformed JSON") from exc
|
|
if isinstance(value, list):
|
|
if allow_profile_list:
|
|
return value
|
|
raise IntakeError("lark-cli returned an unexpected JSON array")
|
|
if not isinstance(value, dict):
|
|
raise IntakeError("lark-cli returned an invalid JSON response")
|
|
if "ok" in value and value["ok"] is not True:
|
|
raise IntakeError("lark-cli returned an error response")
|
|
if "code" in value and (not isinstance(value["code"], int) or isinstance(value["code"], bool) or value["code"] != 0):
|
|
raise IntakeError("lark-cli returned an error response")
|
|
if "ok" not in value and "code" not in value:
|
|
raise IntakeError("lark-cli returned an ambiguous JSON response")
|
|
return value
|
|
|
|
|
|
def profile_check(config: dict[str, Any]) -> None:
|
|
# `profile list` is the official non-mutating profile inspection command.
|
|
value = run_cli(["profile", "list"], allow_profile_list=True)
|
|
if isinstance(value, list):
|
|
profiles = value
|
|
else:
|
|
# Compatibility wrapper: only a successful envelope with a direct list
|
|
# is accepted. Do not loosen this into arbitrary nested objects.
|
|
profiles = value.get("data")
|
|
if not isinstance(profiles, list):
|
|
raise IntakeError("profile check returned an invalid response")
|
|
matching_profile: dict[str, Any] | None = None
|
|
for item in profiles:
|
|
# Match the official profile-list item shape. `user` and
|
|
# `tokenStatus` are optional and deliberately never propagated.
|
|
if (
|
|
not isinstance(item, dict)
|
|
or not isinstance(item.get("name"), str)
|
|
or not PROFILE.fullmatch(item["name"])
|
|
or not isinstance(item.get("appId"), str)
|
|
or not isinstance(item.get("brand"), str)
|
|
or not isinstance(item.get("active"), bool)
|
|
):
|
|
raise IntakeError("profile check returned an invalid profile entry")
|
|
if item["name"] == config["profile"]:
|
|
matching_profile = item
|
|
if matching_profile is None:
|
|
raise IntakeError("configured lark-cli profile does not exist")
|
|
if matching_profile["brand"] != "feishu":
|
|
raise IntakeError("configured lark-cli profile must use the feishu brand")
|
|
|
|
|
|
def text(value: Any) -> str:
|
|
if value is None:
|
|
return ""
|
|
if isinstance(value, str):
|
|
return " ".join(value.split())
|
|
if isinstance(value, float) and not math.isfinite(value):
|
|
raise IntakeError("text field contains a non-finite number")
|
|
if isinstance(value, (int, float, bool)):
|
|
return str(value)
|
|
if isinstance(value, list):
|
|
return "\n".join(part for part in (text(item) for item in value) if part)
|
|
if isinstance(value, dict):
|
|
for key in ("text", "name", "value"):
|
|
if key in value:
|
|
return text(value[key])
|
|
raise IntakeError("text field contains an unsupported object")
|
|
raise IntakeError("text field contains an unsupported value")
|
|
|
|
|
|
def attachment_items(value: Any) -> list[tuple[dict[str, Any], str]]:
|
|
if value in (None, ""):
|
|
return []
|
|
if not isinstance(value, list):
|
|
raise IntakeError("attachments cell must be a list")
|
|
if len(value) > MAX_ATTACHMENTS_PER_RECORD:
|
|
raise IntakeError("record exceeded the attachment count limit")
|
|
attachments: list[tuple[dict[str, Any], str]] = []
|
|
for item in value:
|
|
if not isinstance(item, dict):
|
|
raise IntakeError("attachment metadata must be an object")
|
|
token = item.get("file_token", item.get("token"))
|
|
if not isinstance(token, str) or not SAFE_VALUE.fullmatch(token):
|
|
raise IntakeError("attachment token is invalid")
|
|
metadata = {"name": text(item.get("name")), "type": text(item.get("type", item.get("mime_type"))), "size": item.get("size")}
|
|
if (
|
|
not isinstance(metadata["size"], int)
|
|
or isinstance(metadata["size"], bool)
|
|
or metadata["size"] < 0
|
|
or metadata["size"] > MAX_ATTACHMENT_BYTES
|
|
):
|
|
raise IntakeError("attachment size is invalid")
|
|
attachments.append((metadata, token))
|
|
return attachments
|
|
|
|
|
|
def matrix_from_response(response: dict[str, Any], field_ids: list[str]) -> tuple[list[str], list[list[Any]]]:
|
|
data = response.get("data", response)
|
|
if not isinstance(data, dict):
|
|
raise IntakeError("record list data is invalid")
|
|
fields = data.get("fields")
|
|
ids = data.get("record_id_list", data.get("recordIds"))
|
|
rows = data.get("data", data.get("records", data.get("items", data.get("rows"))))
|
|
if not isinstance(fields, list) or not all(isinstance(x, str) for x in fields):
|
|
raise IntakeError("record list fields are invalid")
|
|
if len(set(fields)) != len(fields) or len(set(field_ids)) != len(field_ids):
|
|
raise IntakeError("record list field projection contains duplicates")
|
|
if set(fields) != set(field_ids):
|
|
raise IntakeError(
|
|
"record list fields do not match configured projection: "
|
|
f"expected={field_ids!r}, actual={fields!r}"
|
|
)
|
|
if not isinstance(ids, list) or not all(isinstance(x, str) and RECORD_ID.fullmatch(x) for x in ids):
|
|
raise IntakeError("record list record_id_list is invalid")
|
|
if not isinstance(rows, list) or len(rows) != len(ids) or any(not isinstance(row, list) or len(row) != len(fields) for row in rows):
|
|
raise IntakeError("record list matrix does not match fields and record_id_list")
|
|
if fields == field_ids:
|
|
return ids, rows
|
|
positions = {field: index for index, field in enumerate(fields)}
|
|
return ids, [
|
|
[row[positions[field_id]] for field_id in field_ids]
|
|
for row in rows
|
|
]
|
|
|
|
|
|
def fetch_pages(config: dict[str, Any]) -> list[tuple[str, list[Any]]]:
|
|
field_order = (
|
|
CLARIFIED_FIELD_ORDER
|
|
if config.get("workflow") == "clarified-writeback-v1"
|
|
else LEGACY_FIELD_ORDER
|
|
)
|
|
logical_fields = [
|
|
logical for logical in field_order if logical in config["fields"]
|
|
]
|
|
field_ids = [config["fields"][logical] for logical in logical_fields]
|
|
all_rows: list[tuple[str, list[Any]]] = []
|
|
offset = 0
|
|
for _ in range(MAX_PAGES):
|
|
args = ["base", "+record-list", "--profile", config["profile"], "--base-token", config["baseToken"], "--table-id", config["tableId"], "--view-id", config["viewId"], "--format", "json", "--offset", str(offset), "--limit", str(PAGE_SIZE)]
|
|
for field_id in field_ids:
|
|
args.extend(["--field-id", field_id])
|
|
response = run_cli(args)
|
|
ids, rows = matrix_from_response(response, field_ids)
|
|
if len(ids) > PAGE_SIZE:
|
|
raise IntakeError("record list exceeded requested page size")
|
|
all_rows.extend(zip(ids, rows))
|
|
if len(all_rows) > MAX_RECORDS:
|
|
raise IntakeError("record list exceeded record limit")
|
|
data = response.get("data", response)
|
|
has_more = data.get("has_more", data.get("hasMore", False))
|
|
if not isinstance(has_more, bool):
|
|
raise IntakeError("record list pagination marker is invalid")
|
|
if not has_more:
|
|
return all_rows
|
|
if not ids:
|
|
raise IntakeError("record list pagination made no progress")
|
|
offset += len(ids)
|
|
raise IntakeError("record list exceeded page limit")
|
|
|
|
|
|
def download(
|
|
config: dict[str, Any], record_id: str, file_token: str,
|
|
output_dir: Path, expected_size: int, timeout: float,
|
|
) -> str:
|
|
try:
|
|
output_dir.mkdir(parents=True, exist_ok=False)
|
|
except OSError as exc:
|
|
raise IntakeError("attachment output directory is unsafe") from exc
|
|
if not output_dir.is_dir() or output_dir.is_symlink():
|
|
raise IntakeError("attachment output directory is unsafe")
|
|
run_cli(
|
|
["base", "+record-download-attachment", "--profile", config["profile"], "--base-token", config["baseToken"], "--table-id", config["tableId"], "--record-id", record_id, "--file-token", file_token, "--output", output_dir.name],
|
|
max_file_bytes=expected_size,
|
|
timeout=timeout,
|
|
cwd=output_dir.parent,
|
|
)
|
|
try:
|
|
created = list(output_dir.iterdir())
|
|
except OSError as exc:
|
|
raise IntakeError("attachment download output is unreadable") from exc
|
|
if len(created) != 1 or not created[0].is_file() or created[0].is_symlink():
|
|
raise IntakeError("attachment download did not produce one safe file")
|
|
root = output_dir.resolve()
|
|
resolved = created[0].resolve()
|
|
if root not in resolved.parents:
|
|
raise IntakeError("attachment download escaped output directory")
|
|
if resolved.stat().st_size != expected_size:
|
|
raise IntakeError("attachment download size did not match metadata")
|
|
return str(resolved)
|
|
|
|
|
|
def source_ref(config: dict[str, Any], record_id: str) -> str:
|
|
"""Return a stable opaque identity without serializing configured identifiers."""
|
|
identity = "\x1f".join(("ack-feishu-base-source-ref-v1", config["profile"], config["baseToken"], config["tableId"], record_id))
|
|
return f"feishu-base:sha256:{hashlib.sha256(identity.encode('utf-8')).hexdigest()}"
|
|
|
|
|
|
def draft_revision(
|
|
record: dict[str, Any], attachment_tokens: list[str] | None = None,
|
|
) -> str:
|
|
"""Bind approval to the normalized source facts and review-controlled fields."""
|
|
tokens = attachment_tokens or []
|
|
if len(tokens) != len(record["attachments"]):
|
|
raise IntakeError("draft revision attachment identity is incomplete")
|
|
stable = {
|
|
key: record[key]
|
|
for key in record
|
|
if key not in {
|
|
"attachments", "warnings", "enrichmentRequired", "draftRevision", "recordId",
|
|
"intakeStatus", "ackTaskId",
|
|
}
|
|
}
|
|
stable["attachments"] = [
|
|
{
|
|
**{key: attachment.get(key) for key in ("name", "type", "size")},
|
|
"tokenDigest": f"sha256:{hashlib.sha256(('ack-feishu-attachment-v1\x1f' + token).encode('utf-8')).hexdigest()}",
|
|
}
|
|
for attachment, token in zip(record["attachments"], tokens)
|
|
]
|
|
encoded = json.dumps(
|
|
stable, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
|
|
|
|
|
|
def fetch(config: dict[str, Any], output_dir: Path | None) -> dict[str, Any]:
|
|
profile_check(config)
|
|
prepared: list[tuple[dict[str, Any], list[tuple[dict[str, Any], str]]]] = []
|
|
batch_warnings: list[dict[str, str]] = []
|
|
total_attachments = 0
|
|
total_attachment_bytes = 0
|
|
for record_id, row in fetch_pages(config):
|
|
workflow = config.get("workflow", "read-only-v1")
|
|
field_order = CLARIFIED_FIELD_ORDER if workflow == "clarified-writeback-v1" else LEGACY_FIELD_ORDER
|
|
logical_fields = [logical for logical in field_order if logical in config["fields"]]
|
|
cells = dict(zip(logical_fields, row))
|
|
attachment_data = attachment_items(cells["attachments"])
|
|
total_attachments += len(attachment_data)
|
|
total_attachment_bytes += sum(metadata["size"] for metadata, _ in attachment_data)
|
|
if total_attachments > MAX_TOTAL_ATTACHMENTS:
|
|
raise IntakeError("batch exceeded the attachment count limit")
|
|
if total_attachment_bytes > MAX_TOTAL_ATTACHMENT_BYTES:
|
|
raise IntakeError("batch exceeded the attachment byte limit")
|
|
if workflow == "clarified-writeback-v1":
|
|
record = {
|
|
"sourceRef": source_ref(config, record_id), "recordId": record_id,
|
|
"updatedAt": text(cells["updatedAt"]), "title": text(cells["title"]),
|
|
"details": text(cells["details"]),
|
|
"problemStatement": text(cells["problemStatement"]),
|
|
"expectedOutcome": text(cells["expectedOutcome"]),
|
|
"acceptance": text(cells["acceptance"]),
|
|
"intakeStatus": text(cells["intakeStatus"]),
|
|
"ackTaskId": text(cells["ackTaskId"]),
|
|
"attachments": [metadata for metadata, _ in attachment_data], "warnings": [],
|
|
}
|
|
else:
|
|
record = {"sourceRef": source_ref(config, record_id), "recordId": record_id, "updatedAt": text(cells["updatedAt"]), "title": text(cells["title"]), "actual": text(cells["actual"]), "expected": text(cells["expected"]), "steps": text(cells["stepsToReproduce"]), "fixLogic": text(cells.get("fixLogic")), "acceptance": text(cells["acceptance"]), "priority": text(cells.get("priority")), "attachments": [metadata for metadata, _ in attachment_data], "warnings": []}
|
|
content_fields = (
|
|
("title", "details", "problemStatement", "expectedOutcome", "acceptance")
|
|
if workflow == "clarified-writeback-v1"
|
|
else BUG_CONTENT_FIELDS
|
|
)
|
|
if not attachment_data and not any(record.get(field) for field in content_fields):
|
|
batch_warnings.append({"recordId": record_id, "code": "blank_record_skipped"})
|
|
continue
|
|
source_fields = ("title", "updatedAt") if workflow == "clarified-writeback-v1" else SOURCE_FACT_FIELDS
|
|
for field in source_fields:
|
|
if not record[field]:
|
|
raise IntakeError(f"record {field} must not be empty")
|
|
enrichment_fields = (["problemStatement", "expectedOutcome", "acceptance"]
|
|
if workflow == "clarified-writeback-v1" else list(COORDINATOR_FIELDS))
|
|
if workflow != "clarified-writeback-v1" and "fixLogic" in config["fields"]:
|
|
enrichment_fields.append("fixLogic")
|
|
record["enrichmentRequired"] = [
|
|
field for field in enrichment_fields if not record[field]
|
|
]
|
|
record["draftRevision"] = draft_revision(
|
|
record, [token for _, token in attachment_data],
|
|
)
|
|
prepared.append((record, attachment_data))
|
|
download_root: Path | None = None
|
|
if output_dir is not None:
|
|
try:
|
|
output_dir.mkdir(parents=True, exist_ok=True)
|
|
if not output_dir.is_dir() or output_dir.is_symlink():
|
|
raise OSError("unsafe output directory")
|
|
download_root = output_dir.resolve(strict=True)
|
|
except OSError as exc:
|
|
raise IntakeError("attachment output directory is unsafe") from exc
|
|
records = []
|
|
attachment_deadline = time.monotonic() + MAX_ATTACHMENT_BATCH_SECONDS
|
|
for record, attachment_data in prepared:
|
|
if download_root is not None:
|
|
for index, (attachment, file_token) in enumerate(attachment_data, start=1):
|
|
remaining = attachment_deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
raise IntakeError("attachment batch exceeded the time limit")
|
|
attachment_dir = download_root / record["recordId"] / f"attachment-{index:02d}"
|
|
attachment["localPath"] = download(
|
|
config, record["recordId"], file_token, attachment_dir,
|
|
attachment["size"], min(60, remaining),
|
|
)
|
|
records.append(record)
|
|
return {"provider": "feishu-base", "workflow": config.get("workflow", "read-only-v1"), "profile": config["profile"], "tableId": config["tableId"], "viewId": config["viewId"], "records": records, "warnings": batch_warnings}
|
|
|
|
|
|
def review_record(
|
|
config: dict[str, Any], record_id: str, expected_source_ref: str,
|
|
expected_revision: str,
|
|
) -> dict[str, Any]:
|
|
"""Resolve one record inside the configured view and bind its reviewed version."""
|
|
if RECORD_ID.fullmatch(record_id) is None:
|
|
raise IntakeError("review record id is invalid")
|
|
if SOURCE_REF.fullmatch(expected_source_ref) is None:
|
|
raise IntakeError("expected source reference is invalid")
|
|
if DRAFT_REVISION.fullmatch(expected_revision) is None:
|
|
raise IntakeError("expected draft revision is invalid")
|
|
matching = [
|
|
record for record in fetch(config, None)["records"]
|
|
if record["recordId"] == record_id
|
|
]
|
|
if len(matching) != 1:
|
|
raise IntakeError("configured review view did not contain exactly one record")
|
|
if (
|
|
matching[0]["sourceRef"] != expected_source_ref
|
|
or matching[0]["draftRevision"] != expected_revision
|
|
):
|
|
raise IntakeError("review record changed before the requested operation")
|
|
return matching[0]
|
|
|
|
|
|
def write_draft(
|
|
config: dict[str, Any], record_id: str, expected_source_ref: str,
|
|
expected_revision: str, draft_path: Path,
|
|
) -> dict[str, Any]:
|
|
"""Overwrite only the review-owned cells for the configured workflow."""
|
|
workflow = config.get("workflow", "read-only-v1")
|
|
if workflow not in {"reviewed-writeback-v1", "clarified-writeback-v1"}:
|
|
raise IntakeError("draft writeback requires a writeback workflow")
|
|
if workflow == "reviewed-writeback-v1" and "fixLogic" not in config["fields"]:
|
|
raise IntakeError("bugIntake.fields.fixLogic is required for draft writeback")
|
|
review_record(config, record_id, expected_source_ref, expected_revision)
|
|
draft = load_draft(draft_path, workflow)
|
|
if workflow == "clarified-writeback-v1":
|
|
patch = {
|
|
config["fields"]["problemStatement"]: draft["problemStatement"],
|
|
config["fields"]["expectedOutcome"]: draft["expectedOutcome"],
|
|
config["fields"]["acceptance"]: "\n".join(
|
|
f"{index}. {item}" for index, item in enumerate(draft["acceptance"], start=1)
|
|
),
|
|
config["fields"]["intakeStatus"]: "待审核",
|
|
}
|
|
written = ["problemStatement", "expectedOutcome", "acceptance", "intakeStatus"]
|
|
else:
|
|
patch = {
|
|
config["fields"]["fixLogic"]: draft["fixLogic"],
|
|
config["fields"]["acceptance"]: "\n".join(
|
|
f"{index}. {item}" for index, item in enumerate(draft["acceptance"], start=1)
|
|
),
|
|
}
|
|
written = ["fixLogic", "acceptance"]
|
|
profile_check(config)
|
|
run_cli([
|
|
"base", "+record-upsert", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--record-id", record_id, "--json",
|
|
json.dumps(patch, ensure_ascii=False, separators=(",", ":")),
|
|
"--format", "json",
|
|
])
|
|
matching = [record for record in fetch(config, None)["records"] if record["recordId"] == record_id]
|
|
if len(matching) != 1 or matching[0]["sourceRef"] != expected_source_ref:
|
|
raise IntakeError("draft writeback readback did not find exactly one record")
|
|
expected_acceptance = text(patch[config["fields"]["acceptance"]])
|
|
if workflow == "clarified-writeback-v1":
|
|
matched = (
|
|
matching[0]["problemStatement"] == text(draft["problemStatement"])
|
|
and matching[0]["expectedOutcome"] == text(draft["expectedOutcome"])
|
|
and matching[0]["acceptance"] == expected_acceptance
|
|
and matching[0]["intakeStatus"] == "待审核"
|
|
)
|
|
else:
|
|
matched = matching[0]["fixLogic"] == text(draft["fixLogic"]) and matching[0]["acceptance"] == expected_acceptance
|
|
if not matched:
|
|
raise IntakeError("draft writeback readback did not match the submitted draft")
|
|
return {
|
|
"provider": "feishu-base",
|
|
"recordId": record_id,
|
|
"written": written,
|
|
"draftRevision": matching[0]["draftRevision"],
|
|
"ok": True,
|
|
}
|
|
|
|
|
|
def import_approved(
|
|
config: dict[str, Any], record_id: str, expected_source_ref: str,
|
|
expected_revision: str,
|
|
) -> dict[str, Any]:
|
|
"""Emit the canonical task payload for one explicitly approved draft revision."""
|
|
workflow = config.get("workflow", "read-only-v1")
|
|
if workflow not in {"reviewed-writeback-v1", "clarified-writeback-v1"}:
|
|
raise IntakeError("approved import requires a writeback workflow")
|
|
record = review_record(
|
|
config, record_id, expected_source_ref, expected_revision,
|
|
)
|
|
acceptance = review_items(record["acceptance"])
|
|
if workflow == "clarified-writeback-v1":
|
|
if record["intakeStatus"] != "已确认":
|
|
raise IntakeError("approved record must have intakeStatus 已确认")
|
|
if not record["problemStatement"] or not record["expectedOutcome"] or not acceptance:
|
|
raise IntakeError("approved record is missing prepared clarification fields")
|
|
task_draft = clarified_task_draft(record)
|
|
return {"provider": "feishu-base", "recordId": record_id, "draftRevision": record["draftRevision"], "taskDraft": task_draft, "ok": True}
|
|
steps = review_items(record["steps"])
|
|
if not record["priority"] or not steps or not record["fixLogic"] or not acceptance:
|
|
raise IntakeError("approved record is missing prepared review fields")
|
|
task_draft: dict[str, Any] = {
|
|
"title": record["title"],
|
|
"priority": record["priority"],
|
|
"description": record["title"],
|
|
"actual": record["actual"],
|
|
"expected": record["expected"],
|
|
"stepsToReproduce": steps,
|
|
"fixLogic": record["fixLogic"],
|
|
"acceptanceCriteria": acceptance,
|
|
"source": {
|
|
"kind": "feishu-base",
|
|
"workflow": "reviewed-writeback-v1",
|
|
"ref": record["sourceRef"],
|
|
"recordId": record["recordId"],
|
|
"updatedAt": record["updatedAt"],
|
|
"approvedRevision": record["draftRevision"],
|
|
},
|
|
}
|
|
task_draft["source"]["approvedPayloadHash"] = approval_payload_hash(task_draft)
|
|
return {
|
|
"provider": "feishu-base",
|
|
"recordId": record_id,
|
|
"draftRevision": record["draftRevision"],
|
|
"taskDraft": task_draft,
|
|
"ok": True,
|
|
}
|
|
|
|
|
|
def clarified_task_draft(record: dict[str, Any]) -> dict[str, Any]:
|
|
"""Map one normalized clarified record to its immutable reviewed task fields."""
|
|
task_draft: dict[str, Any] = {
|
|
"title": record["title"],
|
|
"description": record["problemStatement"],
|
|
"actual": record["details"] or record["title"],
|
|
"expected": record["expectedOutcome"],
|
|
"acceptanceCriteria": review_items(record["acceptance"]),
|
|
"source": {
|
|
"kind": "feishu-base",
|
|
"workflow": "clarified-writeback-v1",
|
|
"ref": record["sourceRef"],
|
|
"recordId": record["recordId"],
|
|
"updatedAt": record["updatedAt"],
|
|
"approvedRevision": record["draftRevision"],
|
|
},
|
|
}
|
|
task_draft["source"]["approvedPayloadHash"] = approval_payload_hash(task_draft)
|
|
return task_draft
|
|
|
|
|
|
def mark_imported(
|
|
board: dict[str, Any], config: dict[str, Any], record_id: str,
|
|
expected_source_ref: str, expected_revision: str, task_id: str,
|
|
) -> dict[str, Any]:
|
|
"""Bind a confirmed Base record to the validated ACK task created from it."""
|
|
if config.get("workflow") != "clarified-writeback-v1":
|
|
raise IntakeError("mark-imported requires clarified-writeback-v1 workflow")
|
|
if not isinstance(task_id, str) or not task_id:
|
|
raise IntakeError("ACK task id is invalid")
|
|
if validate_task_board(board):
|
|
raise IntakeError("task board is invalid for import finalization")
|
|
record = review_record(config, record_id, expected_source_ref, expected_revision)
|
|
|
|
tasks = board.get("tasks")
|
|
if not isinstance(tasks, list):
|
|
raise IntakeError("task board tasks must be a list")
|
|
matches = [
|
|
task for task in tasks
|
|
if isinstance(task, dict) and task.get("id") == task_id
|
|
]
|
|
if len(matches) != 1:
|
|
raise IntakeError("task board did not contain exactly one imported ACK task")
|
|
task = matches[0]
|
|
source = task.get("source")
|
|
expected_task = clarified_task_draft(record)
|
|
reviewed_fields = (
|
|
"title", "description", "actual", "expected", "acceptanceCriteria",
|
|
)
|
|
if (
|
|
not isinstance(source, dict)
|
|
or source != expected_task["source"]
|
|
or any(task.get(field) != expected_task[field] for field in reviewed_fields)
|
|
):
|
|
raise IntakeError("ACK task does not match the approved Base record")
|
|
|
|
if record["intakeStatus"] == "已导入" and record["ackTaskId"] == task_id:
|
|
return {
|
|
"provider": "feishu-base", "recordId": record_id,
|
|
"intakeStatus": "已导入", "ackTaskId": task_id,
|
|
"draftRevision": expected_revision, "ok": True,
|
|
}
|
|
if record["intakeStatus"] != "已确认" or record["ackTaskId"]:
|
|
raise IntakeError("Base record is not ready to mark as imported")
|
|
|
|
profile_check(config)
|
|
run_cli([
|
|
"base", "+record-upsert", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--record-id", record_id, "--json",
|
|
json.dumps({
|
|
config["fields"]["intakeStatus"]: "已导入",
|
|
config["fields"]["ackTaskId"]: task_id,
|
|
}, ensure_ascii=False, separators=(",", ":")),
|
|
"--format", "json",
|
|
])
|
|
matching = [
|
|
item for item in fetch(config, None)["records"]
|
|
if item["recordId"] == record_id
|
|
]
|
|
if (
|
|
len(matching) != 1
|
|
or matching[0]["sourceRef"] != expected_source_ref
|
|
or matching[0]["draftRevision"] != expected_revision
|
|
or matching[0]["intakeStatus"] != "已导入"
|
|
or matching[0]["ackTaskId"] != task_id
|
|
):
|
|
raise IntakeError("import marker readback did not match the ACK task")
|
|
return {
|
|
"provider": "feishu-base", "recordId": record_id,
|
|
"intakeStatus": "已导入", "ackTaskId": task_id,
|
|
"draftRevision": expected_revision, "ok": True,
|
|
}
|
|
|
|
|
|
def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
|
"""Plan idempotent Coordinator actions without mutating the task board."""
|
|
tasks = board.get("tasks")
|
|
if not isinstance(tasks, list):
|
|
raise IntakeError("task board tasks must be a list")
|
|
existing: dict[str, dict[str, Any]] = {}
|
|
for task in tasks:
|
|
if not isinstance(task, dict):
|
|
continue
|
|
source = task.get("source")
|
|
if not isinstance(source, dict) or source.get("kind") != "feishu-base":
|
|
continue
|
|
ref = source.get("ref")
|
|
task_id = task.get("id")
|
|
status = task.get("status")
|
|
updated_at = source.get("updatedAt")
|
|
if (
|
|
not isinstance(ref, str) or SOURCE_REF.fullmatch(ref) is None
|
|
or not isinstance(task_id, str) or not task_id
|
|
or not isinstance(status, str) or not status
|
|
or not isinstance(updated_at, str) or not updated_at
|
|
):
|
|
raise IntakeError("existing Feishu task source is invalid")
|
|
if ref in existing:
|
|
raise IntakeError("task board contains duplicate Feishu source references")
|
|
existing[ref] = task
|
|
|
|
actions: list[dict[str, Any]] = []
|
|
seen_records: set[str] = set()
|
|
for record in records:
|
|
ref = record.get("sourceRef")
|
|
record_id = record.get("recordId")
|
|
updated_at = record.get("updatedAt")
|
|
revision = record.get("draftRevision")
|
|
if (
|
|
not isinstance(ref, str) or SOURCE_REF.fullmatch(ref) is None
|
|
or not isinstance(record_id, str) or RECORD_ID.fullmatch(record_id) is None
|
|
or not isinstance(updated_at, str) or not updated_at
|
|
or not isinstance(revision, str) or DRAFT_REVISION.fullmatch(revision) is None
|
|
):
|
|
raise IntakeError("normalized Feishu record identity is invalid")
|
|
if ref in seen_records:
|
|
raise IntakeError("fetched records contain a duplicate source reference")
|
|
seen_records.add(ref)
|
|
task = existing.get(ref)
|
|
if task is None:
|
|
planned_action: dict[str, Any] = {"sourceRef": ref, "recordId": record_id, "draftRevision": record["draftRevision"], "action": "create"}
|
|
enrichment_required = record.get("enrichmentRequired")
|
|
if enrichment_required:
|
|
planned_action["enrichmentRequired"] = enrichment_required
|
|
actions.append(planned_action)
|
|
continue
|
|
source = task["source"]
|
|
approved_revision = source.get("approvedRevision")
|
|
if isinstance(approved_revision, str):
|
|
if approved_revision == revision:
|
|
action_name = "unchanged"
|
|
elif task["status"] == "open":
|
|
action_name = "refresh"
|
|
else:
|
|
action_name = "drift"
|
|
elif source["updatedAt"] == updated_at:
|
|
action_name = "unchanged"
|
|
elif task["status"] == "open":
|
|
action_name = "refresh"
|
|
else:
|
|
action_name = "drift"
|
|
planned_action = {
|
|
"sourceRef": ref,
|
|
"recordId": record_id,
|
|
"draftRevision": record["draftRevision"],
|
|
"taskId": task["id"],
|
|
"status": task["status"],
|
|
"action": action_name,
|
|
}
|
|
enrichment_required = record.get("enrichmentRequired")
|
|
if enrichment_required and action_name == "refresh":
|
|
planned_action["enrichmentRequired"] = enrichment_required
|
|
actions.append(planned_action)
|
|
return actions
|
|
|
|
|
|
def field_list(config: dict[str, Any]) -> list[dict[str, str]]:
|
|
"""Return the bounded Base field inventory used by schema migration."""
|
|
profile_check(config)
|
|
response = run_cli([
|
|
"base", "+field-list", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--format", "json",
|
|
])
|
|
data = response.get("data", response)
|
|
items = data.get("fields") if isinstance(data, dict) else None
|
|
if not isinstance(items, list) or len(items) > 256:
|
|
raise IntakeError("field list returned an invalid response")
|
|
result: list[dict[str, str]] = []
|
|
for item in items:
|
|
if not isinstance(item, dict) or not all(
|
|
isinstance(item.get(key), str) and item[key]
|
|
for key in ("id", "name", "type")
|
|
):
|
|
raise IntakeError("field list returned an invalid field")
|
|
result.append({key: item[key] for key in ("id", "name", "type")})
|
|
if len({item["name"] for item in result}) != len(result):
|
|
raise IntakeError("field list contains duplicate names")
|
|
return result
|
|
|
|
|
|
def schema_target(config: dict[str, Any]) -> dict[str, str]:
|
|
token_digest = hashlib.sha256(
|
|
("ack-feishu-schema-target-v1\x1f" + config["baseToken"]).encode("utf-8")
|
|
).hexdigest()
|
|
return {
|
|
"profile": config["profile"],
|
|
"baseTokenDigest": f"sha256:{token_digest}",
|
|
"tableId": config["tableId"],
|
|
"viewId": config["viewId"],
|
|
}
|
|
|
|
|
|
def schema_fingerprint(config: dict[str, Any], fields: list[dict[str, str]]) -> str:
|
|
encoded = json.dumps(
|
|
{
|
|
"contract": "clarified-writeback-v1",
|
|
"target": schema_target(config),
|
|
"fields": sorted(fields, key=lambda item: item["id"]),
|
|
},
|
|
ensure_ascii=False, sort_keys=True, separators=(",", ":"),
|
|
).encode("utf-8")
|
|
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
|
|
|
|
|
|
def schema_plan(config: dict[str, Any]) -> dict[str, Any]:
|
|
if config.get("workflow") != "clarified-writeback-v1":
|
|
raise IntakeError("schema migration requires clarified-writeback-v1 workflow")
|
|
fields = field_list(config)
|
|
by_name = {item["name"]: item for item in fields}
|
|
missing = [name for name, _ in TARGET_BASE_FIELDS if name not in by_name]
|
|
wrong_type = [
|
|
{"name": name, "expected": field_type, "actual": by_name[name]["type"]}
|
|
for name, field_type in TARGET_BASE_FIELDS
|
|
if name in by_name and by_name[name]["type"] != field_type
|
|
]
|
|
return {
|
|
"provider": "feishu-base",
|
|
"target": schema_target(config),
|
|
"schemaFingerprint": schema_fingerprint(config, fields),
|
|
"missingFields": missing,
|
|
"typeConflicts": wrong_type,
|
|
"visibleFields": [name for name, _ in TARGET_BASE_FIELDS],
|
|
"legacyFieldsPreserved": [
|
|
name for name in ("期望结果", "问题澄清", "复现步骤", "ACK Ready")
|
|
if name in by_name
|
|
],
|
|
"ok": not wrong_type,
|
|
}
|
|
|
|
|
|
def create_target_field(config: dict[str, Any], name: str) -> None:
|
|
field_type = dict(TARGET_BASE_FIELDS)[name]
|
|
if field_type in {"attachment", "updated_at"}:
|
|
raise IntakeError("schema migration cannot create a missing system/source field")
|
|
payload: dict[str, Any] = {"name": name, "type": field_type}
|
|
if name == "处理状态":
|
|
payload.update({"multiple": False, "options": [{"name": value} for value in INTAKE_STATUSES]})
|
|
run_cli([
|
|
"base", "+field-create", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--json", json.dumps(payload, ensure_ascii=False, separators=(",", ":")),
|
|
"--format", "json",
|
|
])
|
|
|
|
|
|
def migration_rows(config: dict[str, Any], fields: list[dict[str, str]]) -> list[tuple[str, dict[str, Any]]]:
|
|
by_name = {item["name"]: item for item in fields}
|
|
names = ["标题", "详细描述", "期望结果", "处理状态"]
|
|
present = [name for name in names if name in by_name]
|
|
# record-list projects cells by configured field name and returns those
|
|
# names in its matrix, even when the REST field inventory exposes IDs.
|
|
ids = present
|
|
args = [
|
|
"base", "+record-list", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--view-id", config["viewId"], "--format", "json", "--offset", "0",
|
|
"--limit", str(PAGE_SIZE),
|
|
]
|
|
for field_id in ids:
|
|
args.extend(["--field-id", field_id])
|
|
response = run_cli(args)
|
|
record_ids, rows = matrix_from_response(response, ids)
|
|
data = response.get("data", response)
|
|
if data.get("has_more", data.get("hasMore", False)):
|
|
raise IntakeError("schema migration view exceeded one bounded page")
|
|
return [(record_id, dict(zip(present, row))) for record_id, row in zip(record_ids, rows)]
|
|
|
|
|
|
def schema_apply(config: dict[str, Any], expected_fingerprint: str) -> dict[str, Any]:
|
|
if DRAFT_REVISION.fullmatch(expected_fingerprint) is None:
|
|
raise IntakeError("expected schema fingerprint is invalid")
|
|
before = schema_plan(config)
|
|
if before["schemaFingerprint"] != expected_fingerprint:
|
|
raise IntakeError("Base schema changed after planning")
|
|
if before["typeConflicts"]:
|
|
raise IntakeError("Base schema has incompatible target field types")
|
|
for name in before["missingFields"]:
|
|
create_target_field(config, name)
|
|
fields = field_list(config)
|
|
for _ in range(4):
|
|
if all(name in {item["name"] for item in fields} for name, _ in TARGET_BASE_FIELDS):
|
|
break
|
|
time.sleep(0.5)
|
|
fields = field_list(config)
|
|
by_name = {item["name"]: item for item in fields}
|
|
if any(name not in by_name for name, _ in TARGET_BASE_FIELDS):
|
|
raise IntakeError("schema migration did not create all target fields")
|
|
migrated: list[str] = []
|
|
for record_id, cells in migration_rows(config, fields):
|
|
details = text(cells.get("详细描述"))
|
|
legacy_expected = text(cells.get("期望结果"))
|
|
patch: dict[str, Any] = {}
|
|
marker = f"用户原始期望：{legacy_expected}" if legacy_expected else ""
|
|
if marker and marker not in details:
|
|
patch["详细描述"] = f"{details}\n\n{marker}".strip()
|
|
if not text(cells.get("处理状态")):
|
|
patch["处理状态"] = "待整理"
|
|
if patch:
|
|
run_cli([
|
|
"base", "+record-upsert", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--record-id", record_id, "--json",
|
|
json.dumps(patch, ensure_ascii=False, separators=(",", ":")),
|
|
"--format", "json",
|
|
])
|
|
migrated.append(record_id)
|
|
visible_ids = [by_name[name]["id"] for name, _ in TARGET_BASE_FIELDS]
|
|
run_cli([
|
|
"base", "+view-set-visible-fields", "--profile", config["profile"],
|
|
"--base-token", config["baseToken"], "--table-id", config["tableId"],
|
|
"--view-id", config["viewId"], "--json",
|
|
json.dumps({"visible_fields": visible_ids}, separators=(",", ":")),
|
|
"--format", "json",
|
|
])
|
|
return {
|
|
"provider": "feishu-base", "createdFields": before["missingFields"],
|
|
"migratedRecordIds": migrated, "visibleFields": [name for name, _ in TARGET_BASE_FIELDS],
|
|
"schemaFingerprint": schema_fingerprint(config, fields), "ok": True,
|
|
}
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description="Read and review a configured Feishu Base bug intake")
|
|
sub = parser.add_subparsers(dest="command", required=True)
|
|
for name in ("check", "fetch", "plan", "write-draft", "import-approved", "mark-imported", "schema-plan", "schema-apply"):
|
|
command = sub.add_parser(name)
|
|
command.add_argument("tasks", type=Path, help="ACK tasks.yaml or JSON board")
|
|
if name in {"fetch", "plan"}:
|
|
command.add_argument("--output-dir", type=Path, help="explicit directory for downloaded attachments")
|
|
if name in {"write-draft", "import-approved", "mark-imported"}:
|
|
command.add_argument("--record-id", required=True, help="existing Feishu Base record id")
|
|
command.add_argument("--expected-source-ref", required=True, help="sourceRef returned by fetch")
|
|
command.add_argument("--expected-draft-revision", required=True, help="draftRevision returned by fetch")
|
|
if name == "write-draft":
|
|
command.add_argument("--input", type=Path, required=True, help="bounded JSON draft file")
|
|
if name == "mark-imported":
|
|
command.add_argument("--task-id", required=True, help="validated ACK task id")
|
|
if name == "schema-apply":
|
|
command.add_argument("--expected-schema-fingerprint", required=True)
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
board = load_board(args.tasks)
|
|
config = config_from_board(board)
|
|
if args.command == "check":
|
|
profile_check(config)
|
|
output = {"provider": "feishu-base", "workflow": config.get("workflow", "read-only-v1"), "profile": config["profile"], "ok": True}
|
|
elif args.command == "fetch":
|
|
output = fetch(config, args.output_dir)
|
|
elif args.command == "plan":
|
|
output = fetch(config, args.output_dir)
|
|
output["actions"] = plan_actions(board, output["records"])
|
|
elif args.command == "schema-plan":
|
|
output = schema_plan(config)
|
|
elif args.command == "schema-apply":
|
|
output = schema_apply(config, args.expected_schema_fingerprint)
|
|
elif args.command == "write-draft":
|
|
output = write_draft(
|
|
config, args.record_id, args.expected_source_ref,
|
|
args.expected_draft_revision, args.input,
|
|
)
|
|
elif args.command == "mark-imported":
|
|
output = mark_imported(
|
|
board, config, args.record_id, args.expected_source_ref,
|
|
args.expected_draft_revision, args.task_id,
|
|
)
|
|
else:
|
|
output = import_approved(
|
|
config, args.record_id, args.expected_source_ref,
|
|
args.expected_draft_revision,
|
|
)
|
|
except IntakeError as exc:
|
|
sys.stderr.write(f"Feishu bug intake failed: {exc}\n")
|
|
return 1
|
|
except (OSError, subprocess.SubprocessError):
|
|
sys.stderr.write("Feishu bug intake failed: local I/O failed\n")
|
|
return 1
|
|
print(json.dumps(output, ensure_ascii=False, separators=(",", ":")))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|