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())