335 lines
14 KiB
Python
335 lines
14 KiB
Python
#!/usr/bin/env python3
|
||
"""Create a read-only 1C change proposal from a task evidence bundle."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import base64
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
def decode_arg(value: str | None, encoded: str | None) -> str | None:
|
||
if encoded:
|
||
return base64.b64decode(encoded).decode("utf-8")
|
||
return value
|
||
|
||
|
||
def load_json(path: Path) -> dict[str, Any]:
|
||
return json.loads(path.read_text(encoding="utf-8-sig"))
|
||
|
||
|
||
def norm(value: Any) -> str:
|
||
return re.sub(r"[\s._-]+", "", str(value or "")).casefold()
|
||
|
||
|
||
def words(text: str) -> set[str]:
|
||
return set(re.findall(r"[A-Za-zА-Яа-яЁё0-9_]{3,}", text.casefold()))
|
||
|
||
|
||
def infer_intents(text: str) -> list[str]:
|
||
w = words(text)
|
||
intents = []
|
||
if w & {"добавить", "создать", "вывести", "показать"}:
|
||
intents.append("add_or_expose")
|
||
if w & {"изменить", "исправить", "доработать", "поменять"}:
|
||
intents.append("modify")
|
||
if w & {"проверить", "проверка", "проанализировать"}:
|
||
intents.append("inspect_or_verify")
|
||
if w & {"кнопка", "кнопку", "команда", "команду"}:
|
||
intents.append("form_command")
|
||
if w & {"форма", "форму"}:
|
||
intents.append("form")
|
||
if w & {"реквизит", "реквизиты", "поле", "поля"}:
|
||
intents.append("attribute")
|
||
if w & {"проведение", "провести", "запись", "записать"}:
|
||
intents.append("object_lifecycle")
|
||
return intents or ["inspect_or_verify"]
|
||
|
||
|
||
def unique_list(values: list[Any]) -> list[Any]:
|
||
result = []
|
||
seen = set()
|
||
for value in values:
|
||
key = json.dumps(value, ensure_ascii=False, sort_keys=True)
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
result.append(value)
|
||
return result
|
||
|
||
|
||
def collect_search_findings(investigation: dict[str, Any]) -> list[dict[str, Any]]:
|
||
findings = []
|
||
for search in investigation.get("searches") or []:
|
||
for match in search.get("matches") or []:
|
||
evidence = match.get("evidence") or {}
|
||
findings.append(
|
||
{
|
||
"search_text": search.get("text"),
|
||
"area": match.get("area"),
|
||
"name": match.get("name"),
|
||
"title": match.get("title"),
|
||
"form": match.get("form"),
|
||
"origin": match.get("origin"),
|
||
"effective_action": match.get("effective_action"),
|
||
"path": evidence.get("path") or evidence.get("form_xml_path"),
|
||
"line": evidence.get("line"),
|
||
"id": evidence.get("id"),
|
||
"text": evidence.get("text"),
|
||
}
|
||
)
|
||
return unique_list(findings)
|
||
|
||
|
||
def collect_file_targets(investigation: dict[str, Any]) -> list[dict[str, Any]]:
|
||
targets = []
|
||
for form_context in investigation.get("forms") or []:
|
||
for form in form_context.get("forms") or []:
|
||
targets.append(
|
||
{
|
||
"kind": "form_xml",
|
||
"name": form.get("name"),
|
||
"origin": form.get("origin"),
|
||
"path": form.get("form_xml_path"),
|
||
"module_path": form.get("module_path"),
|
||
"reason": "selected_form_context",
|
||
}
|
||
)
|
||
for overlay in form.get("extension_overlays") or []:
|
||
targets.append(
|
||
{
|
||
"kind": "form_xml",
|
||
"name": overlay.get("name"),
|
||
"origin": overlay.get("origin"),
|
||
"path": overlay.get("form_xml_path"),
|
||
"module_path": overlay.get("module_path"),
|
||
"reason": "active_extension_form_overlay",
|
||
}
|
||
)
|
||
for module_context in investigation.get("modules") or []:
|
||
for module in module_context.get("modules") or []:
|
||
targets.append(
|
||
{
|
||
"kind": "bsl_module",
|
||
"name": module.get("name"),
|
||
"origin": module.get("origin"),
|
||
"path": module.get("path"),
|
||
"relative_path": module.get("relative_path"),
|
||
"reason": "selected_module_context",
|
||
}
|
||
)
|
||
for snippet in investigation.get("code_snippets") or []:
|
||
targets.append(
|
||
{
|
||
"kind": "bsl_module",
|
||
"name": snippet.get("module"),
|
||
"origin": snippet.get("origin"),
|
||
"path": snippet.get("path"),
|
||
"line": snippet.get("focus_line"),
|
||
"reason": f"code_search_hit:{snippet.get('search_text')}",
|
||
}
|
||
)
|
||
return unique_list([target for target in targets if target.get("path") or target.get("module_path")])
|
||
|
||
|
||
def extension_names_from_targets(targets: list[dict[str, Any]]) -> list[str]:
|
||
names = []
|
||
for target in targets:
|
||
origin = target.get("origin") or {}
|
||
if origin.get("layer") == "extension" and origin.get("extension"):
|
||
names.append(origin["extension"])
|
||
return sorted(set(names))
|
||
|
||
|
||
def choose_write_strategy(investigation: dict[str, Any], targets: list[dict[str, Any]]) -> dict[str, Any]:
|
||
extension_names = extension_names_from_targets(targets)
|
||
active_extensions = ((investigation.get("brief") or {}).get("active_extensions") or [])
|
||
if extension_names:
|
||
preferred = extension_names[0]
|
||
reason = "task evidence already points to an active extension overlay"
|
||
elif active_extensions:
|
||
preferred = active_extensions[0]
|
||
reason = "active extensions exist; extension-first write path is required"
|
||
else:
|
||
preferred = None
|
||
reason = "no active extension target found; create a new extension in a disposable validation loop"
|
||
return {
|
||
"mode": "extension_first_proposal",
|
||
"preferred_extension": preferred,
|
||
"reason": reason,
|
||
"forbidden": [
|
||
"direct SQL metadata/data updates",
|
||
"direct Config/ConfigSave/ConfigCAS writes",
|
||
"automatic production Designer update/apply",
|
||
],
|
||
}
|
||
|
||
|
||
def split_write_targets(targets: list[dict[str, Any]], preferred_extension: str | None) -> dict[str, list[dict[str, Any]]]:
|
||
write_candidates = []
|
||
references = []
|
||
for target in targets:
|
||
origin = target.get("origin") or {}
|
||
if origin.get("layer") == "extension" and (not preferred_extension or origin.get("extension") == preferred_extension):
|
||
write_candidates.append(target)
|
||
else:
|
||
references.append(target)
|
||
return {
|
||
"write_candidates": unique_list(write_candidates),
|
||
"read_only_reference_files": unique_list(references),
|
||
}
|
||
|
||
|
||
def summarize_existing_state(findings: list[dict[str, Any]], targets: list[dict[str, Any]]) -> dict[str, Any]:
|
||
form_commands = [row for row in findings if row.get("area") in {"form.command", "form.item"}]
|
||
code_hits = [row for row in findings if row.get("area") == "module.code"]
|
||
attributes = [row for row in findings if row.get("area") == "metadata.attribute"]
|
||
return {
|
||
"found_form_commands_or_items": form_commands,
|
||
"found_metadata_attributes": attributes,
|
||
"found_code_hits": code_hits[:20],
|
||
"target_files": targets,
|
||
}
|
||
|
||
|
||
def build_steps(task_text: str, intents: list[str], investigation: dict[str, Any], findings: list[dict[str, Any]], targets: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
object_info = ((investigation.get("brief") or {}).get("object") or {})
|
||
steps = [
|
||
{
|
||
"order": 1,
|
||
"action": "confirm_target_object",
|
||
"object": object_info,
|
||
"evidence": investigation.get("candidate"),
|
||
},
|
||
{
|
||
"order": 2,
|
||
"action": "review_effective_metadata",
|
||
"purpose": "ensure extension-added attributes and modified types are included before code generation",
|
||
},
|
||
]
|
||
next_order = 3
|
||
if "form_command" in intents or any(row.get("area") in {"form.command", "form.item"} for row in findings):
|
||
steps.append(
|
||
{
|
||
"order": next_order,
|
||
"action": "review_form_command",
|
||
"purpose": "determine whether the command already exists or must be added",
|
||
"evidence": [row for row in findings if row.get("area") in {"form.command", "form.item"}],
|
||
}
|
||
)
|
||
next_order += 1
|
||
if any(row.get("area") == "module.code" for row in findings):
|
||
steps.append(
|
||
{
|
||
"order": next_order,
|
||
"action": "review_related_bsl",
|
||
"purpose": "inspect code snippets before proposing BSL changes",
|
||
"evidence": [row for row in findings if row.get("area") == "module.code"][:10],
|
||
}
|
||
)
|
||
next_order += 1
|
||
steps.append(
|
||
{
|
||
"order": next_order,
|
||
"action": "draft_extension_patch",
|
||
"purpose": "prepare a reviewable patch against extension source files only",
|
||
"candidate_targets": targets[:20],
|
||
}
|
||
)
|
||
steps.append(
|
||
{
|
||
"order": next_order + 1,
|
||
"action": "validate_in_disposable_base",
|
||
"purpose": "load/update extension, open affected forms, run smoke scenario, and only then ask for approval",
|
||
}
|
||
)
|
||
return steps
|
||
|
||
|
||
def build_checks(intents: list[str]) -> list[dict[str, Any]]:
|
||
checks = [
|
||
{"name": "syntax_check", "description": "Run 1C/BSL syntax validation for changed modules."},
|
||
{"name": "extension_load", "description": "Package/load the extension into a disposable base."},
|
||
{"name": "form_open", "description": "Open affected forms and verify command bindings and handlers."},
|
||
]
|
||
if "form_command" in intents:
|
||
checks.append({"name": "command_invocation", "description": "Invoke the affected form command and verify expected behavior."})
|
||
if "object_lifecycle" in intents:
|
||
checks.append({"name": "write_posting_smoke", "description": "Create/write/post a test document in a disposable base."})
|
||
return checks
|
||
|
||
|
||
def proposal_for_investigation(task_text: str, investigation: dict[str, Any]) -> dict[str, Any]:
|
||
intents = infer_intents(task_text)
|
||
findings = collect_search_findings(investigation)
|
||
targets = collect_file_targets(investigation)
|
||
strategy = choose_write_strategy(investigation, targets)
|
||
target_split = split_write_targets(targets, strategy.get("preferred_extension"))
|
||
return {
|
||
"object": (investigation.get("brief") or {}).get("object"),
|
||
"candidate": investigation.get("candidate"),
|
||
"intents": intents,
|
||
"write_strategy": strategy,
|
||
"existing_state": summarize_existing_state(findings, targets),
|
||
"target_policy": target_split,
|
||
"implementation_steps": build_steps(task_text, intents, investigation, findings, targets),
|
||
"validation_checks": build_checks(intents),
|
||
"open_questions": infer_open_questions(task_text, findings),
|
||
}
|
||
|
||
|
||
def infer_open_questions(task_text: str, findings: list[dict[str, Any]]) -> list[str]:
|
||
questions = []
|
||
if "добав" in task_text.casefold() and any(row.get("area") in {"form.command", "form.item"} for row in findings):
|
||
questions.append("The requested command/item appears to already exist; confirm whether the task is to modify existing behavior rather than add a duplicate.")
|
||
if not findings:
|
||
questions.append("No direct matches were found in the evidence bundle; broaden search terms or inspect more objects before generating a patch.")
|
||
return questions
|
||
|
||
|
||
def build_proposal(evidence: dict[str, Any]) -> dict[str, Any]:
|
||
task_text = ((evidence.get("task") or {}).get("text") or "")
|
||
proposals = [proposal_for_investigation(task_text, item) for item in evidence.get("investigations") or []]
|
||
return {
|
||
"schema": "onec_task_change_proposal.v1",
|
||
"task": evidence.get("task"),
|
||
"source_evidence_schema": evidence.get("schema"),
|
||
"view": evidence.get("view"),
|
||
"proposals": proposals,
|
||
"safety": {
|
||
"mode": "proposal_only",
|
||
"write_status": "blocked_until_write_gates",
|
||
"write_contract": "docs/1c-write-path-safety.md",
|
||
},
|
||
"counts": {
|
||
"proposals": len(proposals),
|
||
"target_files": sum(len((proposal.get("existing_state") or {}).get("target_files") or []) for proposal in proposals),
|
||
"write_candidates": sum(len((proposal.get("target_policy") or {}).get("write_candidates") or []) for proposal in proposals),
|
||
"open_questions": sum(len(proposal.get("open_questions") or []) for proposal in proposals),
|
||
},
|
||
}
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Create a read-only 1C change proposal from evidence.")
|
||
parser.add_argument("--evidence", type=Path, required=True)
|
||
parser.add_argument("--output", type=Path)
|
||
args = parser.parse_args()
|
||
|
||
result = build_proposal(load_json(args.evidence))
|
||
output = json.dumps(result, ensure_ascii=False, indent=2) + "\n"
|
||
if args.output:
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
args.output.write_text(output, encoding="utf-8")
|
||
print(json.dumps({"output": str(args.output), "counts": result["counts"]}, ensure_ascii=False))
|
||
else:
|
||
print(output)
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|