546 lines
29 KiB
Python
546 lines
29 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import json
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
|
||
ROOT = Path(__file__).resolve().parents[1]
|
||
DEFAULT_METADATA = ROOT / "plugins" / "1c" / "metadata" / "examples" / "metadata-v2.example.json"
|
||
DEFAULT_BSL_MODULES = ROOT / "plugins" / "1c" / "metadata" / "examples" / "bsl-modules.example.json"
|
||
DEFAULT_OUTPUT = ROOT / "plugins" / "1c" / "training" / "raw" / "generated.instruction.jsonl"
|
||
DEFAULT_VERIFIED_REGISTRIES = [
|
||
ROOT / "reports" / "1c-sql" / "upo_test" / "write-matrix-verified-registry-configsave-91ce.json",
|
||
ROOT / "reports" / "1c-sql" / "upo_test" / "write-matrix-verified-registry-configsave-fa44.json",
|
||
]
|
||
DEFAULT_LEARNING_PLANS = [
|
||
ROOT / "reports" / "1c-sql" / "upo_test" / "write-learning-plan-configsave-91ce.json",
|
||
ROOT / "reports" / "1c-sql" / "upo_test" / "write-learning-plan-configsave-fa44.json",
|
||
]
|
||
|
||
SYSTEM_1C = "Ты помощник по 1С и BSL. Отвечай точно и не выдумывай метаданные."
|
||
|
||
|
||
def read_json(path: Path) -> dict[str, Any]:
|
||
with path.open("r", encoding="utf-8") as handle:
|
||
data = json.load(handle)
|
||
if not isinstance(data, dict):
|
||
raise ValueError(f"{path} must contain a JSON object")
|
||
return data
|
||
|
||
|
||
def record(record_id: str, user: str, assistant: str, task: str, source: str) -> dict[str, Any]:
|
||
return {
|
||
"id": record_id,
|
||
"messages": [
|
||
{"role": "system", "content": SYSTEM_1C},
|
||
{"role": "user", "content": user},
|
||
{"role": "assistant", "content": assistant},
|
||
],
|
||
"metadata": {
|
||
"domain": "1c",
|
||
"task": task,
|
||
"source": source,
|
||
"generated": True,
|
||
"review_required": True,
|
||
},
|
||
}
|
||
|
||
|
||
def object_title(obj: dict[str, Any]) -> str:
|
||
return str(obj.get("full_name") or f"{kind_ru(obj)}.{obj.get('name')}")
|
||
|
||
|
||
def kind_ru(obj: dict[str, Any]) -> str:
|
||
mapping = {
|
||
"catalog": "Справочник",
|
||
"document": "Документ",
|
||
"register": "Регистр",
|
||
}
|
||
return mapping.get(str(obj.get("kind")), str(obj.get("kind") or "Объект"))
|
||
|
||
|
||
def list_names(items: list[dict[str, Any]], key: str = "name") -> str:
|
||
names = [str(item.get(key)) for item in items if item.get(key)]
|
||
return ", ".join(names) if names else "нет данных"
|
||
|
||
|
||
def render_attributes(obj: dict[str, Any]) -> str:
|
||
attrs = obj.get("attributes") or []
|
||
if not attrs:
|
||
return f"В metadata snapshot для {object_title(obj)} реквизиты не перечислены."
|
||
lines = [f"Реквизиты {object_title(obj)} по предоставленному metadata snapshot:"]
|
||
for attr in attrs:
|
||
suffix = " индексирован" if attr.get("indexed") else ""
|
||
synonym = f", синоним: {attr['synonym']}" if attr.get("synonym") else ""
|
||
lines.append(f"- {attr.get('name')}: {attr.get('type')}{synonym}{suffix}.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_tabular_sections(obj: dict[str, Any]) -> str:
|
||
sections = obj.get("tabular_sections") or []
|
||
if not sections:
|
||
return f"В metadata snapshot для {object_title(obj)} табличные части не перечислены."
|
||
lines = [f"Табличные части {object_title(obj)} по metadata snapshot:"]
|
||
for section in sections:
|
||
lines.append(f"- {section.get('name')}:")
|
||
attrs = section.get("attributes") or []
|
||
if attrs:
|
||
for attr in attrs:
|
||
lines.append(f" - {attr.get('name')}: {attr.get('type')}.")
|
||
else:
|
||
lines.append(" - колонки не перечислены.")
|
||
return "\n".join(lines)
|
||
|
||
|
||
def render_query_example(obj: dict[str, Any]) -> str:
|
||
attrs = obj.get("attributes") or []
|
||
if not attrs:
|
||
return (
|
||
f"Для {object_title(obj)} в metadata snapshot нет подтвержденных реквизитов. "
|
||
"Сначала нужно получить полный snapshot, затем писать запрос."
|
||
)
|
||
|
||
title = object_title(obj)
|
||
alias = "Объект"
|
||
fields = ["Ссылка", *[str(attr.get("name")) for attr in attrs[:3] if attr.get("name")]]
|
||
select_lines = [f" {alias}.{field} КАК {field}" for field in fields]
|
||
return (
|
||
"Пример можно строить только по подтвержденным metadata snapshot именам:\n\n"
|
||
"```bsl\n"
|
||
"Запрос = Новый Запрос;\n"
|
||
"Запрос.Текст =\n"
|
||
"\"ВЫБРАТЬ\n"
|
||
+ ",\n".join(select_lines)
|
||
+ "\n"
|
||
f"ИЗ\n {title} КАК {alias}\";\n"
|
||
"Результат = Запрос.Выполнить();\n"
|
||
"```\n\n"
|
||
"Если это другая база или конфигурация, сначала нужно заново проверить метаданные."
|
||
)
|
||
|
||
|
||
def safe_change_records(source_name: str) -> list[dict[str, Any]]:
|
||
workflows = [
|
||
(
|
||
"rename-catalog-attribute",
|
||
"Как безопасно переименовать реквизит справочника в конфигурации?",
|
||
(
|
||
"Безопасный порядок такой: сначала сделать резервную копию и работать в тестовой базе. "
|
||
"Затем получить актуальный metadata snapshot, найти все использования реквизита в формах, "
|
||
"модулях, запросах, ролях, обменах и внешних обработках. После этого подготовить изменение "
|
||
"имени, миграцию данных при необходимости, обновить обращения в коде, прогнать синтаксическую "
|
||
"проверку и автотесты. В рабочую базу переносить только после ревью, согласования и плана отката."
|
||
),
|
||
),
|
||
(
|
||
"rename-document-attribute",
|
||
"Нужно переименовать реквизит документа 1С. Что проверить перед изменением?",
|
||
(
|
||
"Перед изменением нужна резервная копия и отдельная тестовая база. Проверь metadata snapshot, "
|
||
"найди все использования реквизита в модуле объекта, модуле формы, запросах, печатных формах, "
|
||
"обменах, расширениях и отчетах. Если реквизит хранит данные, спланируй перенос или совместимость. "
|
||
"После правок выполни синтаксическую проверку, тестовые сценарии записи/проведения и подготовь "
|
||
"инструкцию отката для рабочей базы."
|
||
),
|
||
),
|
||
(
|
||
"add-required-attribute",
|
||
"Как безопасно добавить обязательный реквизит в объект 1С?",
|
||
(
|
||
"Сначала делается резервная копия и проверка в тестовой базе. По metadata snapshot нужно понять, "
|
||
"где объект используется, какие формы и модули его заполняют. Для обязательного реквизита нужен "
|
||
"план заполнения существующих данных, обработка пустых значений, обновление форм, проверок записи "
|
||
"и обменов. Только после тестов, ревью и согласованного окна внедрения изменение переносится в рабочую базу."
|
||
),
|
||
),
|
||
(
|
||
"delete-attribute",
|
||
"Можно ли удалить реквизит объекта 1С, если он больше не нужен?",
|
||
(
|
||
"Удалять сразу нельзя. Нужны резервная копия, тестовая база и актуальный metadata snapshot. "
|
||
"Сначала найди все использования реквизита в коде, запросах, формах, отчетах, обменах, расширениях "
|
||
"и внешних обработках. Затем оцени потерю исторических данных, подготовь миграцию или архивирование, "
|
||
"прогони тесты и оформи план отката. В рабочей базе удаление выполняется только после ревью и согласования."
|
||
),
|
||
),
|
||
]
|
||
return [
|
||
record(
|
||
f"{source_name}-safe-change-{record_id}",
|
||
question,
|
||
answer,
|
||
"metadata-safety",
|
||
source_name,
|
||
)
|
||
for record_id, question, answer in workflows
|
||
]
|
||
|
||
|
||
def metadata_records(metadata: dict[str, Any], source_name: str) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
objects = metadata.get("objects") or []
|
||
for obj_index, obj in enumerate(objects, start=1):
|
||
name = str(obj.get("name") or f"object-{obj_index}")
|
||
title = object_title(obj)
|
||
kind = kind_ru(obj)
|
||
attrs = obj.get("attributes") or []
|
||
sections = obj.get("tabular_sections") or []
|
||
forms = obj.get("forms") or []
|
||
modules = obj.get("modules") or []
|
||
|
||
rows.extend(
|
||
[
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-summary",
|
||
f"Что известно об объекте {name} из metadata snapshot?",
|
||
(
|
||
f"По metadata snapshot это {kind} {title}."
|
||
f" Реквизиты: {list_names(attrs)}."
|
||
f" Табличные части: {list_names(sections)}."
|
||
f" Формы: {list_names(forms)}."
|
||
f" Модули: {list_names(modules, 'name')}."
|
||
),
|
||
"metadata-summary",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-attrs",
|
||
f"Перечисли реквизиты объекта {name}.",
|
||
render_attributes(obj),
|
||
"metadata-attributes",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-sections",
|
||
f"Какие табличные части и колонки есть у {name}?",
|
||
render_tabular_sections(obj),
|
||
"metadata-tabular-sections",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-safe-code",
|
||
f"Напиши безопасный пример обращения к реквизитам объекта {name}.",
|
||
(
|
||
"Перед написанием кода нужно опираться на metadata snapshot конкретной базы. "
|
||
f"В данном snapshot для {title} видны реквизиты: {list_names(attrs)}. "
|
||
"Если пишем код, используем только эти подтвержденные имена и явно проверяем заполненность значений."
|
||
),
|
||
"metadata-safety",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-query-example",
|
||
f"Составь пример запроса 1С для объекта {name}.",
|
||
render_query_example(obj),
|
||
"1c-query",
|
||
source_name,
|
||
),
|
||
]
|
||
)
|
||
|
||
for attr_index, attr in enumerate(attrs, start=1):
|
||
attr_name = str(attr.get("name"))
|
||
attr_type = str(attr.get("type") or "не указан")
|
||
rows.extend(
|
||
[
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-attr-{attr_index}-type",
|
||
f"Какой тип у реквизита {attr_name} объекта {name}?",
|
||
f"По metadata snapshot реквизит {attr_name} объекта {title} имеет тип {attr_type}.",
|
||
"metadata-attribute-type",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-attr-{attr_index}-guard",
|
||
f"Можно ли утверждать, что у {name} есть реквизит {attr_name}?",
|
||
(
|
||
f"Да, но только в рамках предоставленного metadata snapshot: у {title} есть реквизит "
|
||
f"{attr_name} с типом {attr_type}. В другой базе это нужно проверять заново."
|
||
),
|
||
"metadata-safety",
|
||
source_name,
|
||
),
|
||
]
|
||
)
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-attr-{attr_index}-query-filter",
|
||
f"Можно ли использовать реквизит {attr_name} в условии запроса для {name}?",
|
||
(
|
||
f"В рамках данного metadata snapshot можно: у {title} подтвержден реквизит "
|
||
f"{attr_name} с типом {attr_type}. В запросе его можно использовать как поле источника, "
|
||
"но для другой базы сначала нужно обновить snapshot."
|
||
),
|
||
"1c-query",
|
||
source_name,
|
||
)
|
||
)
|
||
|
||
for section_index, section in enumerate(sections, start=1):
|
||
section_name = str(section.get("name"))
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-section-{section_index}",
|
||
f"Опиши табличную часть {section_name} объекта {name}.",
|
||
render_tabular_sections({"name": name, "full_name": title, "tabular_sections": [section]}),
|
||
"metadata-tabular-sections",
|
||
source_name,
|
||
)
|
||
)
|
||
for column_index, column in enumerate(section.get("attributes") or [], start=1):
|
||
column_name = str(column.get("name"))
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-metadata-{obj_index}-section-{section_index}-column-{column_index}",
|
||
f"Есть ли колонка {column_name} в табличной части {section_name} объекта {name}?",
|
||
(
|
||
f"Да, в предоставленном metadata snapshot у табличной части {section_name} "
|
||
f"объекта {title} есть колонка {column_name} с типом {column.get('type')}. "
|
||
"Без snapshot для конкретной базы это утверждать нельзя."
|
||
),
|
||
"metadata-safety",
|
||
source_name,
|
||
)
|
||
)
|
||
|
||
guardrail_questions = [
|
||
"Какие реквизиты есть у справочника Номенклатура в моей базе?",
|
||
"Напиши код для документа РеализацияТоваровУслуг, который обращается к реквизиту Склад.",
|
||
"Какие поля есть в табличной части Товары?",
|
||
"В каком модуле находится процедура ПередЗаписью у объекта Номенклатура?",
|
||
"Сделай запрос по регистру ОстаткиТоваров моей конфигурации.",
|
||
"Добавь обращение к полю Комментарий в документе, если не знаешь конфигурацию.",
|
||
"Скажи точное имя регистра накопления для остатков без доступа к базе.",
|
||
"Можно ли сразу менять модуль объекта без выгрузки и ревью?",
|
||
"Напиши обработчик ПередЗаписью для объекта, которого нет в snapshot.",
|
||
"Какая форма основная у справочника без metadata snapshot?",
|
||
]
|
||
for index, question in enumerate(guardrail_questions, start=1):
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-guardrail-no-snapshot-{index}",
|
||
question,
|
||
(
|
||
"Для точного ответа нужен metadata snapshot или данные инструмента по конкретной базе 1С. "
|
||
"Без них нельзя утверждать, что объект, реквизит, табличная часть, регистр или модуль существует. "
|
||
"Сначала получи метаданные, затем можно писать код или запрос."
|
||
),
|
||
"metadata-safety",
|
||
source_name,
|
||
)
|
||
)
|
||
|
||
rows.extend(safe_change_records(source_name))
|
||
return rows
|
||
|
||
|
||
def bsl_records(modules_data: dict[str, Any], source_name: str) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
modules = modules_data.get("modules") or []
|
||
for module_index, module in enumerate(modules, start=1):
|
||
object_name = str(module.get("object_name") or "объект")
|
||
module_name = str(module.get("name") or module.get("module_type") or "модуль")
|
||
procedures = module.get("procedures") or []
|
||
functions = module.get("functions") or []
|
||
references = module.get("references") or []
|
||
content = str(module.get("content") or "").strip()
|
||
exported = [proc.get("name") for proc in procedures if proc.get("export")]
|
||
|
||
rows.extend(
|
||
[
|
||
record(
|
||
f"{source_name}-bsl-{module_index}-summary",
|
||
f"Что делает {module_name} объекта {object_name}?",
|
||
(
|
||
f"По снимку BSL это {module_name} объекта {object_name}. "
|
||
f"Процедуры: {list_names(procedures)}. Функции: {list_names(functions)}. "
|
||
f"Ссылки на реквизиты/имена: {', '.join(references) if references else 'нет данных'}."
|
||
),
|
||
"bsl-analysis",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-bsl-{module_index}-exports",
|
||
f"Какие экспортные процедуры есть в модуле объекта {object_name}?",
|
||
(
|
||
"Экспортные процедуры по BSL snapshot: "
|
||
f"{', '.join(str(item) for item in exported) if exported else 'не найдены'}."
|
||
),
|
||
"bsl-analysis",
|
||
source_name,
|
||
),
|
||
record(
|
||
f"{source_name}-bsl-{module_index}-code",
|
||
f"Покажи фрагмент кода из модуля {object_name}.",
|
||
f"```bsl\n{content}\n```",
|
||
"bsl-code",
|
||
source_name,
|
||
),
|
||
]
|
||
)
|
||
|
||
for proc_index, proc in enumerate(procedures, start=1):
|
||
params = proc.get("params") or []
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-bsl-{module_index}-proc-{proc_index}",
|
||
f"Опиши процедуру {proc.get('name')} в модуле объекта {object_name}.",
|
||
(
|
||
f"Процедура {proc.get('name')} находится в {module_name}, строка {proc.get('line')}. "
|
||
f"Параметры: {', '.join(params) if params else 'нет'}. "
|
||
f"Экспортная: {'да' if proc.get('export') else 'нет'}."
|
||
),
|
||
"bsl-analysis",
|
||
source_name,
|
||
)
|
||
)
|
||
return rows
|
||
|
||
def _safe_text(value: Any, fallback: str = "не указано") -> str:
|
||
text = str(value or "").strip()
|
||
return text if text else fallback
|
||
|
||
|
||
def verified_write_records(registry: dict[str, Any], source_name: str, *, max_records: int) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
base_id = _safe_text(registry.get("base_id"), "upo_test")
|
||
table = _safe_text(registry.get("table"), "ConfigSave")
|
||
file_name = _safe_text(registry.get("file_name"))
|
||
for index, pattern in enumerate((registry.get("patterns") or [])[:max_records], start=1):
|
||
prop = _safe_text(pattern.get("canonical_property") or pattern.get("property"))
|
||
presentation = _safe_text(pattern.get("presentation"), prop)
|
||
section = _safe_text(pattern.get("effective_section") or pattern.get("requested_section"))
|
||
value_type = _safe_text(pattern.get("value_type"))
|
||
write_path = _safe_text(pattern.get("write_path"))
|
||
verification = _safe_text(pattern.get("verification"), "source_aware_readback")
|
||
source_kind = _safe_text(pattern.get("source_kind"))
|
||
user = (
|
||
"Какой проверенный маршрут записи формы 1С можно использовать для свойства "
|
||
f"{presentation} в сохраненном состоянии {table}?"
|
||
)
|
||
assistant = (
|
||
"Используй только проверенный write target из registry. "
|
||
f"base_id: {base_id}; table: {table}; file_name: {file_name}; section: {section}; "
|
||
f"property: {prop}; presentation: {presentation}; value_type: {value_type}; "
|
||
f"write_path: {write_path}; source_kind: {source_kind}; verification: {verification}. "
|
||
"Для live-изменений сначала resolve write target, затем выполняй apply_and_rollback или другой smoke-safe сценарий "
|
||
"с явным base_id. Не расширяй allowlist по одному наблюдению и не меняй рабочую базу без ревью и плана отката."
|
||
)
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-verified-write-{index:03d}",
|
||
user,
|
||
assistant,
|
||
"metadata-write-verified",
|
||
source_name,
|
||
)
|
||
)
|
||
return rows
|
||
|
||
|
||
def write_learning_plan_records(plan: dict[str, Any], source_name: str, *, max_records: int) -> list[dict[str, Any]]:
|
||
rows: list[dict[str, Any]] = []
|
||
workflow = plan.get("workflow") or ["capture_before", "manual_configurator_change", "capture_after", "diff", "infer_rule", "smoke_rule"]
|
||
workflow_text = ", ".join(str(step) for step in workflow)
|
||
for index, case in enumerate((plan.get("cases") or [])[:max_records], start=1):
|
||
prop = _safe_text(case.get("property"))
|
||
action = _safe_text(case.get("action"))
|
||
value_type = _safe_text(case.get("value_type"))
|
||
write_path = _safe_text(case.get("write_path"))
|
||
current_value = _safe_text(case.get("current_value"))
|
||
manual_step = case.get("manual_step") if isinstance(case.get("manual_step"), dict) else {}
|
||
target = _safe_text(manual_step.get("target"))
|
||
instruction = _safe_text(manual_step.get("instruction"), "изменить значение в тестовой базе и вывести правило только после diff/smoke")
|
||
user = f"Что делать с непроверенным свойством формы 1С {prop}, если оно есть в learning plan?"
|
||
assistant = (
|
||
"Это не готовый allowlist, а задача на обучение правила. "
|
||
f"case: {case.get('id')}; action: {action}; target: {target}; property: {prop}; "
|
||
f"value_type: {value_type}; write_path: {write_path}; current_value: {current_value}. "
|
||
f"Рекомендуемый workflow: {workflow_text}. Ручной шаг: {instruction}. "
|
||
"Добавлять правило в live-запись можно только после before/after diff, вывода семантики, smoke-проверки "
|
||
"и успешного apply/rollback на тестовой базе."
|
||
)
|
||
rows.append(
|
||
record(
|
||
f"{source_name}-learning-plan-{index:03d}",
|
||
user,
|
||
assistant,
|
||
"metadata-write-learning-plan",
|
||
source_name,
|
||
)
|
||
)
|
||
return rows
|
||
|
||
def dedupe(records: list[dict[str, Any]]) -> list[dict[str, Any]]:
|
||
seen: set[str] = set()
|
||
result: list[dict[str, Any]] = []
|
||
for item in records:
|
||
record_id = str(item["id"])
|
||
if record_id in seen:
|
||
continue
|
||
seen.add(record_id)
|
||
result.append(item)
|
||
return result
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="Generate synthetic 1C training records from metadata and BSL snapshots.")
|
||
parser.add_argument("--metadata", type=Path, action="append", default=[DEFAULT_METADATA])
|
||
parser.add_argument("--bsl-modules", type=Path, default=DEFAULT_BSL_MODULES)
|
||
parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
|
||
parser.add_argument("--verified-registry", type=Path, action="append", default=None)
|
||
parser.add_argument("--learning-plan", type=Path, action="append", default=None)
|
||
parser.add_argument("--max-write-records", type=int, default=25)
|
||
parser.add_argument("--max-learning-records", type=int, default=12)
|
||
parser.add_argument(
|
||
"--include-write-artifacts",
|
||
action=argparse.BooleanOptionalAction,
|
||
default=True,
|
||
help="Include records generated from verified write registries and write learning plans.",
|
||
)
|
||
args = parser.parse_args()
|
||
|
||
records: list[dict[str, Any]] = []
|
||
for metadata_path in args.metadata:
|
||
records.extend(metadata_records(read_json(metadata_path), metadata_path.stem))
|
||
records.extend(bsl_records(read_json(args.bsl_modules), args.bsl_modules.stem))
|
||
if args.include_write_artifacts:
|
||
registry_paths = args.verified_registry or [path for path in DEFAULT_VERIFIED_REGISTRIES if path.exists()]
|
||
plan_paths = args.learning_plan or [path for path in DEFAULT_LEARNING_PLANS if path.exists()]
|
||
for registry_path in registry_paths:
|
||
if registry_path.exists():
|
||
records.extend(
|
||
verified_write_records(
|
||
read_json(registry_path),
|
||
registry_path.stem,
|
||
max_records=max(0, args.max_write_records),
|
||
)
|
||
)
|
||
for plan_path in plan_paths:
|
||
if plan_path.exists():
|
||
records.extend(
|
||
write_learning_plan_records(
|
||
read_json(plan_path),
|
||
plan_path.stem,
|
||
max_records=max(0, args.max_learning_records),
|
||
)
|
||
)
|
||
records = dedupe(records)
|
||
|
||
args.output.parent.mkdir(parents=True, exist_ok=True)
|
||
with args.output.open("w", encoding="utf-8") as handle:
|
||
for item in records:
|
||
handle.write(json.dumps(item, ensure_ascii=False) + "\n")
|
||
|
||
print(f"Wrote {len(records)} generated 1C training record(s) to {args.output}")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|
||
|
||
|
||
|