247 lines
9.3 KiB
Python
247 lines
9.3 KiB
Python
#!/usr/bin/env python3
|
||
"""Safely append or replace one BSL routine inside a 1C patch workspace."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from validate_1c_patch_workspace_semantics import validate_workspace
|
||
|
||
|
||
WORD = r"А-Яа-яA-Za-z0-9_"
|
||
ROUTINE_RE = re.compile(rf"(?<![{WORD}])(Процедура|Функция)\s+([А-Яа-яA-Za-z_][{WORD}]*)", re.IGNORECASE)
|
||
END_RE = {
|
||
"процедура": re.compile(rf"(?<![{WORD}])КонецПроцедуры(?![{WORD}])", re.IGNORECASE),
|
||
"функция": re.compile(rf"(?<![{WORD}])КонецФункции(?![{WORD}])", re.IGNORECASE),
|
||
}
|
||
|
||
|
||
def load_json(path: Path) -> dict[str, Any]:
|
||
return json.loads(path.read_text(encoding="utf-8-sig"))
|
||
|
||
|
||
def write_json(path: Path, data: dict[str, Any]) -> None:
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
|
||
|
||
|
||
def read_text(path: Path) -> str:
|
||
try:
|
||
return path.read_text(encoding="utf-8-sig")
|
||
except UnicodeDecodeError:
|
||
return path.read_text(encoding="cp1251", errors="replace")
|
||
|
||
|
||
def normalize(value: str | None) -> str:
|
||
return re.sub(r"[\s._-]+", "", str(value or "")).casefold()
|
||
|
||
|
||
def line_starts(text: str) -> list[int]:
|
||
starts = [0]
|
||
for match in re.finditer(r"\n", text):
|
||
starts.append(match.end())
|
||
return starts
|
||
|
||
|
||
def offset_to_line(starts: list[int], offset: int) -> int:
|
||
line = 1
|
||
for index, start in enumerate(starts, start=1):
|
||
if start > offset:
|
||
break
|
||
line = index
|
||
return line
|
||
|
||
|
||
def routine_blocks(text: str) -> list[dict[str, Any]]:
|
||
starts = line_starts(text)
|
||
blocks = []
|
||
matches = list(ROUTINE_RE.finditer(text))
|
||
for match in matches:
|
||
kind = match.group(1)
|
||
name = match.group(2)
|
||
end_match = END_RE[kind.casefold()].search(text, match.end())
|
||
end = end_match.end() if end_match else len(text)
|
||
blocks.append(
|
||
{
|
||
"kind": kind,
|
||
"name": name,
|
||
"normalized_name": normalize(name),
|
||
"start": match.start(),
|
||
"declaration_start": match.start(),
|
||
"end": end,
|
||
"line_start": offset_to_line(starts, match.start()),
|
||
"line_end": offset_to_line(starts, end),
|
||
}
|
||
)
|
||
return blocks
|
||
|
||
|
||
def directive_start(text: str, declaration_start: int) -> int:
|
||
prefix = text[:declaration_start]
|
||
lines = prefix.splitlines(keepends=True)
|
||
start_offset = len(prefix)
|
||
index = len(lines) - 1
|
||
while index >= 0:
|
||
line = lines[index]
|
||
stripped = line.strip()
|
||
if stripped.startswith("&"):
|
||
start_offset -= len(line)
|
||
index -= 1
|
||
continue
|
||
if stripped == "":
|
||
candidate = index - 1
|
||
while candidate >= 0 and lines[candidate].strip() == "":
|
||
candidate -= 1
|
||
if candidate >= 0 and lines[candidate].strip().startswith("&"):
|
||
start_offset -= len(line)
|
||
index -= 1
|
||
continue
|
||
break
|
||
return start_offset
|
||
|
||
|
||
def one_routine_from_text(routine_text: str) -> dict[str, Any]:
|
||
blocks = routine_blocks(routine_text)
|
||
if len(blocks) != 1:
|
||
raise SystemExit(f"Routine text must contain exactly one procedure/function, found {len(blocks)}.")
|
||
block = blocks[0]
|
||
if block["end"] < len(routine_text.rstrip()):
|
||
suffix = routine_text[block["end"] :].strip()
|
||
if suffix:
|
||
raise SystemExit("Routine text must not contain extra code after the routine end.")
|
||
return block
|
||
|
||
|
||
def decode_routine_text(args: argparse.Namespace) -> str:
|
||
sources = [bool(args.routine_text), bool(args.routine_text_b64), bool(args.routine_file)]
|
||
if sum(sources) != 1:
|
||
raise SystemExit("Use exactly one of --routine-text, --routine-text-b64, or --routine-file.")
|
||
if args.routine_text is not None:
|
||
return args.routine_text
|
||
if args.routine_text_b64:
|
||
return base64.b64decode(args.routine_text_b64).decode("utf-8")
|
||
return Path(args.routine_file).read_text(encoding="utf-8-sig")
|
||
|
||
|
||
def safe_relative(relative_path: str) -> str:
|
||
normalized = relative_path.replace("\\", "/")
|
||
path = Path(normalized)
|
||
if path.is_absolute() or ".." in path.parts or not normalized:
|
||
raise SystemExit(f"Unsafe relative path: {relative_path}")
|
||
return normalized
|
||
|
||
|
||
def manifest_record(workspace: Path, relative_path: str) -> dict[str, Any]:
|
||
manifest = load_json(workspace / "manifest.json")
|
||
wanted = safe_relative(relative_path)
|
||
for record in manifest.get("files") or []:
|
||
record_relative = str(record.get("relative_path") or "").replace("\\", "/")
|
||
if record_relative == wanted:
|
||
if record.get("kind") != "bsl_module":
|
||
raise SystemExit(f"Manifest target is not a BSL module: {relative_path}")
|
||
return record
|
||
raise SystemExit(f"BSL module is not present in workspace manifest: {relative_path}")
|
||
|
||
|
||
def ensure_inside_working(workspace: Path, relative_path: str) -> Path:
|
||
working_root = (workspace / "working").resolve()
|
||
path = (working_root / Path(safe_relative(relative_path))).resolve()
|
||
if not str(path).casefold().startswith(str(working_root).casefold()):
|
||
raise SystemExit(f"Refusing path outside working/: {path}")
|
||
if not path.exists():
|
||
raise SystemExit(f"Working BSL module does not exist: {path}")
|
||
return path
|
||
|
||
|
||
def apply_edit(path: Path, routine_text: str, *, operation: str) -> dict[str, Any]:
|
||
new_block = one_routine_from_text(routine_text)
|
||
text = read_text(path)
|
||
blocks = routine_blocks(text)
|
||
matches = [block for block in blocks if block["normalized_name"] == new_block["normalized_name"]]
|
||
if len(matches) > 1:
|
||
raise SystemExit(f"Target module already has duplicate routine: {new_block['name']}")
|
||
exists = bool(matches)
|
||
if operation == "append" and exists:
|
||
raise SystemExit(f"Routine already exists, append refused: {new_block['name']}")
|
||
if operation == "replace" and not exists:
|
||
raise SystemExit(f"Routine does not exist, replace refused: {new_block['name']}")
|
||
|
||
replacement = routine_text.strip() + "\n"
|
||
if exists:
|
||
old = matches[0]
|
||
start = directive_start(text, int(old["declaration_start"]))
|
||
end = int(old["end"])
|
||
while end < len(text) and text[end : end + 1] in {"\r", "\n"}:
|
||
end += 1
|
||
updated = text[:start].rstrip() + "\n\n" + replacement + text[end:].lstrip("\r\n")
|
||
status = "replaced"
|
||
span = {"old_line_start": old["line_start"], "old_line_end": old["line_end"]}
|
||
else:
|
||
separator = "\n\n" if text.strip() else ""
|
||
updated = text.rstrip() + separator + replacement
|
||
status = "appended"
|
||
span = {}
|
||
|
||
path.write_text(updated, encoding="utf-8")
|
||
return {
|
||
"status": status,
|
||
"routine": {"kind": new_block["kind"], "name": new_block["name"]},
|
||
"path": str(path),
|
||
**span,
|
||
}
|
||
|
||
|
||
def edit_workspace(workspace: Path, relative_path: str, routine_text: str, *, operation: str, keep_on_failure: bool) -> dict[str, Any]:
|
||
record = manifest_record(workspace, relative_path)
|
||
path = ensure_inside_working(workspace, str(record.get("relative_path") or relative_path))
|
||
before = read_text(path)
|
||
edit = apply_edit(path, routine_text, operation=operation)
|
||
semantic = validate_workspace(workspace)
|
||
rolled_back = False
|
||
if not semantic.get("passed") and not keep_on_failure:
|
||
path.write_text(before, encoding="utf-8")
|
||
rolled_back = True
|
||
return {
|
||
"schema": "onec_bsl_routine_edit.v1",
|
||
"workspace": str(workspace),
|
||
"relative_path": str(record.get("relative_path")),
|
||
"operation": operation,
|
||
"edit": edit,
|
||
"rolled_back": rolled_back,
|
||
"semantic_validation": {
|
||
"schema": semantic.get("schema"),
|
||
"passed": semantic.get("passed"),
|
||
"counts": semantic.get("counts"),
|
||
"findings": semantic.get("findings"),
|
||
},
|
||
"passed": bool(semantic.get("passed")) and not rolled_back,
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Append/replace one BSL routine in a 1C patch workspace working file.")
|
||
parser.add_argument("--workspace", type=Path, required=True)
|
||
parser.add_argument("--relative-path", required=True)
|
||
parser.add_argument("--operation", choices=["append", "replace", "upsert"], default="upsert")
|
||
parser.add_argument("--routine-text")
|
||
parser.add_argument("--routine-text-b64")
|
||
parser.add_argument("--routine-file", type=Path)
|
||
parser.add_argument("--keep-on-failure", action="store_true", help="Keep the edit even when semantic validation fails.")
|
||
parser.add_argument("--output", type=Path)
|
||
args = parser.parse_args()
|
||
|
||
result = edit_workspace(args.workspace, args.relative_path, decode_routine_text(args), operation=args.operation, keep_on_failure=args.keep_on_failure)
|
||
if args.output:
|
||
write_json(args.output, result)
|
||
print(json.dumps({"output": str(args.output) if args.output else None, "passed": result["passed"], "rolled_back": result["rolled_back"], "edit": result["edit"], "semantic": result["semantic_validation"]["counts"]}, ensure_ascii=False))
|
||
return 0 if result["passed"] else 2
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|