Prepare AI-ready 1C structure packages
CI / python (push) Has been cancelled
CI / rust (push) Has been cancelled

This commit is contained in:
2026-05-21 20:43:14 +03:00
parent 5f066d2f6b
commit e86f6be385
7 changed files with 475 additions and 0 deletions
@@ -0,0 +1,219 @@
from __future__ import annotations
import json
from pathlib import Path
from typing import Any
from one_c_normalizer import NormalizedProject, normalize_one_c_project
from semantic_kernel import index_project
from sir import SirSnapshot, snapshot_to_json
AI_STRUCTURE_VERSION = "1.0"
_PARSEABLE_SUFFIXES = {".xml", ".mdo", ".bsl"}
_BINARY_1C_SUFFIXES = {".cf", ".cfe"}
def prepare_ai_structure(
*,
project_id: str,
input_path: Path,
output_path: Path,
structure_only: bool = False,
) -> dict[str, Any]:
if not input_path.exists():
raise FileNotFoundError(f"Input path not found: {input_path}")
output_path.mkdir(parents=True, exist_ok=True)
files = _inventory(input_path)
parseable = any(Path(item["relative_path"]).suffix.casefold() in _PARSEABLE_SUFFIXES for item in files)
binaries = [item for item in files if Path(item["relative_path"]).suffix.casefold() in _BINARY_1C_SUFFIXES]
diagnostics: list[str] = []
snapshot: SirSnapshot | None = None
normalized: NormalizedProject | None = None
if parseable:
snapshot = index_project(input_path, project_id=project_id, structure_only=structure_only)
try:
normalized = normalize_one_c_project(input_path, project_id=project_id)
except Exception as error:
diagnostics.append(f"NormalizedProject build failed: {error}")
elif binaries:
diagnostics.append(
"Input contains only binary .cf/.cfe files. Server-side AI structure requires Designer DumpConfigToFiles "
"or Windows Agent export before semantic indexing."
)
else:
diagnostics.append("No 1C metadata/XML/BSL files or .cf/.cfe binaries were found.")
manifest = _manifest(project_id, input_path, output_path, files, snapshot, normalized, diagnostics, binaries)
_write_json(output_path / "manifest.json", manifest)
_write_json(output_path / "source_inventory.json", {"files": files})
if snapshot is not None:
(output_path / "sir_snapshot.json").write_bytes(snapshot_to_json(snapshot))
_write_json(output_path / "ai_objects.json", _ai_objects(snapshot))
_write_json(output_path / "ai_modules.json", _ai_modules(snapshot))
_write_json(output_path / "ai_edges.json", [edge.model_dump(mode="json") for edge in snapshot.edges])
if normalized is not None:
_write_json(output_path / "normalized_project.json", normalized.model_dump(mode="json"))
_write_text(output_path / "ai_context.md", _ai_context_markdown(manifest, snapshot, normalized))
_write_text(output_path / "export_plan.md", _export_plan_markdown(project_id, input_path, output_path, binaries, parseable))
return manifest
def _inventory(root: Path) -> list[dict[str, Any]]:
paths = [root] if root.is_file() else sorted(path for path in root.rglob("*") if path.is_file())
return [
{
"relative_path": path.name if root.is_file() else path.relative_to(root).as_posix(),
"suffix": path.suffix.casefold(),
"size": path.stat().st_size,
}
for path in paths
]
def _manifest(
project_id: str,
input_path: Path,
output_path: Path,
files: list[dict[str, Any]],
snapshot: SirSnapshot | None,
normalized: NormalizedProject | None,
diagnostics: list[str],
binaries: list[dict[str, Any]],
) -> dict[str, Any]:
return {
"version": AI_STRUCTURE_VERSION,
"project_id": project_id,
"input_path": str(input_path),
"output_path": str(output_path),
"status": "ready" if snapshot is not None or normalized is not None else "export_required",
"files_count": len(files),
"binary_1c_files": binaries,
"artifacts": _artifacts(snapshot, normalized),
"snapshot": None
if snapshot is None
else {
"snapshot_id": snapshot.snapshot_id,
"snapshot_hash": snapshot.snapshot_hash,
"nodes": len(snapshot.nodes),
"edges": len(snapshot.edges),
"diagnostics": len(snapshot.diagnostics),
},
"normalized": None
if normalized is None
else {
"objects": sum(len(group.objects) for group in normalized.configuration.groups),
"groups": len(normalized.configuration.groups),
"extensions": len(normalized.configuration.extensions),
"access_profiles": len(normalized.access.profiles),
"access_groups": len(normalized.access.groups),
"access_users": len(normalized.access.users),
},
"diagnostics": diagnostics,
}
def _artifacts(snapshot: SirSnapshot | None, normalized: NormalizedProject | None) -> list[str]:
artifacts = ["manifest.json", "source_inventory.json", "ai_context.md", "export_plan.md"]
if snapshot is not None:
artifacts.extend(["sir_snapshot.json", "ai_objects.json", "ai_modules.json", "ai_edges.json"])
if normalized is not None:
artifacts.append("normalized_project.json")
return artifacts
def _ai_objects(snapshot: SirSnapshot) -> list[dict[str, Any]]:
return [
{
"kind": node.kind.value if hasattr(node.kind, "value") else str(node.kind),
"name": node.name,
"qualified_name": node.qualified_name,
"lineage_id": node.lineage_id,
"semantic_id": node.semantic_id,
"source": None if node.source_ref is None else node.source_ref.model_dump(mode="json"),
"attributes": node.attributes,
}
for node in snapshot.nodes
if (node.kind.value if hasattr(node.kind, "value") else str(node.kind)) != "MODULE"
]
def _ai_modules(snapshot: SirSnapshot) -> list[dict[str, Any]]:
return [
{
"name": node.name,
"qualified_name": node.qualified_name,
"lineage_id": node.lineage_id,
"source": None if node.source_ref is None else node.source_ref.model_dump(mode="json"),
"attributes": node.attributes,
}
for node in snapshot.nodes
if (node.kind.value if hasattr(node.kind, "value") else str(node.kind)) == "MODULE"
]
def _ai_context_markdown(manifest: dict[str, Any], snapshot: SirSnapshot | None, normalized: NormalizedProject | None) -> str:
lines = [
f"# SFERA AI structure: {manifest['project_id']}",
"",
f"- Status: {manifest['status']}",
f"- Source files: {manifest['files_count']}",
f"- Artifacts: {', '.join(manifest['artifacts'])}",
]
if snapshot is not None:
lines.extend(
[
f"- SIR nodes: {len(snapshot.nodes)}",
f"- SIR edges: {len(snapshot.edges)}",
f"- Snapshot hash: {snapshot.snapshot_hash}",
]
)
if normalized is not None:
lines.append(f"- Normalized metadata groups: {len(normalized.configuration.groups)}")
if manifest["diagnostics"]:
lines.append("")
lines.append("## Diagnostics")
lines.extend(f"- {item}" for item in manifest["diagnostics"])
lines.extend(
[
"",
"## How AI should use this package",
"- Use `normalized_project.json` as the authoritative 1C object model.",
"- Use `sir_snapshot.json`, `ai_objects.json`, `ai_modules.json`, and `ai_edges.json` for code navigation and impact analysis.",
"- Treat modules/forms/commands as parts of their owner 1C objects, not as detached text files.",
]
)
return "\n".join(lines) + "\n"
def _export_plan_markdown(project_id: str, input_path: Path, output_path: Path, binaries: list[dict[str, Any]], parseable: bool) -> str:
lines = [
f"# 1C export plan for {project_id}",
"",
f"- Input: `{input_path}`",
f"- Output: `{output_path}`",
]
if parseable:
lines.append("- Metadata files were found; semantic processing was executed directly.")
if binaries:
lines.extend(["", "## Binary .cf/.cfe files", ""])
for item in binaries:
lines.append(f"- `{item['relative_path']}`")
lines.extend(
[
"",
"Для полной структуры выполните экспорт через 1C Designer/Windows Agent:",
"- `/DumpConfigToFiles <output>/configuration -Format Hierarchical` для основной конфигурации",
"- `/DumpConfigToFiles <output>/extensions/<name> -Format Hierarchical -Extension <name>` для расширений",
"- затем повторно запустите подготовку AI-структуры на папке с выгруженными файлами.",
]
)
return "\n".join(lines) + "\n"
def _write_json(path: Path, payload: Any) -> None:
path.write_text(json.dumps(payload, ensure_ascii=False, indent=2, default=str), encoding="utf-8")
def _write_text(path: Path, payload: str) -> None:
path.write_text(payload, encoding="utf-8")
@@ -46,6 +46,7 @@ def _topbar(project_id: str, project_nav: str) -> str:
<a class="button" href="/docs">API</a>
<a class="button" href="/html5/operations">Операции</a>
<a class="button" href="/html5/projects/{quote(project_id)}/access">Права</a>
<a class="button" href="/html5/projects/{quote(project_id)}/ai-structure">AI-структура</a>
<a class="button" href="/html5/projects/{quote(project_id)}/setup">HTML5 Setup</a>
<a class="button" href="/editor?project={quote(project_id)}">Legacy Next</a>
</header>"""
@@ -0,0 +1,93 @@
from __future__ import annotations
from html import escape
from typing import Iterable
from urllib.parse import quote
from api_server.html5 import _page, _project_link, _topbar
def render_html5_ai_structure_page(*, project_id: str, projects: Iterable[object], result: dict | None = None) -> str:
project_nav = "\n".join(_project_link(project, project_id) for project in projects)
return _page(
f"SFERA AI Structure - {project_id}",
f"""
<main class="workspace ai-structure-workspace" data-html5-page="ai-structure" data-project-id="{escape(project_id)}">
{_topbar(project_id, project_nav)}
<section class="setup-layout">
<aside class="panel">
<div class="setup-card">
<p class="eyebrow">AI-ready export</p>
<h1>Структура для ИИ</h1>
<p class="muted">Сервер подготовит полный пакет SFERA: normalized model, SIR graph, объекты, модули, связи и контекст для генерации кода.</p>
</div>
</aside>
<section class="panel setup-main">
<div class="panel-title">Подготовка структуры</div>
{render_html5_ai_structure_form(project_id)}
<div data-html5-ai-structure-result>{render_html5_ai_structure_result(result)}</div>
</section>
</section>
</main>
""",
)
def render_html5_ai_structure_form(project_id: str) -> str:
return f"""
<form
class="ai-structure-form"
hx-post="/html5/projects/{quote(project_id)}/ai-structure/run"
hx-target="[data-html5-ai-structure-result]"
hx-swap="innerHTML"
>
<label>
<span>Папка с cf/cfe или выгрузкой</span>
<input name="input_path" placeholder="D:\\1c\\source" />
</label>
<label>
<span>Папка результата</span>
<input name="output_path" placeholder="D:\\sfera-ai\\result" />
</label>
<label>
<span>Project id</span>
<input name="project_id" value="{escape(project_id)}" />
</label>
<button class="primary" type="submit">Подготовить для ИИ</button>
</form>
"""
def render_html5_ai_structure_result(result: dict | None) -> str:
if result is None:
return '<p class="muted padded">Укажите входную и выходную папку. Файлы будут созданы сервером в указанном каталоге.</p>'
diagnostics = list(result.get("diagnostics") or [])
artifacts = list(result.get("artifacts") or [])
snapshot = result.get("snapshot") or {}
normalized = result.get("normalized") or {}
return f"""
<section class="ai-structure-result" data-html5-ai-structure-status="{escape(str(result.get('status', '')))}">
<div class="access-plan-head">
<span class="status-pill">{escape(str(result.get("status", "")))}</span>
<strong>{escape(str(result.get("output_path", "")))}</strong>
</div>
<dl class="setup-metrics">
<div><dt>Файлы</dt><dd>{escape(str(result.get("files_count", 0)))}</dd></div>
<div><dt>Nodes</dt><dd>{escape(str(snapshot.get("nodes", 0)))}</dd></div>
<div><dt>Edges</dt><dd>{escape(str(snapshot.get("edges", 0)))}</dd></div>
<div><dt>Objects</dt><dd>{escape(str(normalized.get("objects", 0)))}</dd></div>
</dl>
<div class="panel-title">Артефакты</div>
<div class="access-operations">{''.join(f'<article class="access-card"><strong>{escape(str(item))}</strong><small>AI structure package</small></article>' for item in artifacts)}</div>
{_diagnostics(diagnostics)}
</section>
"""
def _diagnostics(items: list[object]) -> str:
if not items:
return ""
return f"""
<div class="panel-title">Диагностика</div>
<ul class="access-warnings">{''.join(f'<li>{escape(str(item))}</li>' for item in items)}</ul>
"""
@@ -0,0 +1,35 @@
from __future__ import annotations
from collections.abc import Callable, Iterable
from pathlib import Path
from typing import Any
from fastapi import HTTPException
from api_server.html5_ai_structure import render_html5_ai_structure_page, render_html5_ai_structure_result
from api_server.html5_forms import form_value
def html5_ai_structure_page(
*,
project_id: str,
project_summaries: Callable[[], Iterable[object]],
) -> str:
return render_html5_ai_structure_page(project_id=project_id, projects=project_summaries())
def html5_ai_structure_run(
*,
project_id: str,
form: dict[str, list[str]],
prepare: Callable[..., dict[str, Any]],
) -> str:
effective_project_id = form_value(form, "project_id") or project_id
input_path = form_value(form, "input_path")
output_path = form_value(form, "output_path")
if not input_path:
raise HTTPException(status_code=400, detail="input_path is required")
if not output_path:
raise HTTPException(status_code=400, detail="output_path is required")
result = prepare(project_id=effective_project_id, input_path=Path(input_path), output_path=Path(output_path))
return render_html5_ai_structure_result(result)
@@ -63,6 +63,11 @@ from api_server.html5_access_controller import (
html5_access_publish_plan as _html5_access_publish_plan,
html5_access_user_detail as _html5_access_user_detail,
)
from api_server.ai_structure_service import prepare_ai_structure as _prepare_ai_structure
from api_server.html5_ai_structure_controller import (
html5_ai_structure_page as _html5_ai_structure_page,
html5_ai_structure_run as _html5_ai_structure_run,
)
from api_server.html5_editor_controller import (
html5_editor_page as _html5_editor_page,
html5_form_editor_page as _html5_form_editor_page,
@@ -840,6 +845,27 @@ class ImportRequest(BaseModel):
mode: ImportMode = ImportMode.FULL_REPLACE
class AiStructurePrepareRequest(BaseModel):
input_path: str
output_path: str
project_id: str | None = None
structure_only: bool = False
class AiStructurePrepareResponse(BaseModel):
version: str
project_id: str
input_path: str
output_path: str
status: str
files_count: int = 0
binary_1c_files: list[dict] = Field(default_factory=list)
artifacts: list[str] = Field(default_factory=list)
snapshot: dict | None = None
normalized: dict | None = None
diagnostics: list[str] = Field(default_factory=list)
class AccessProfileDraftRequest(BaseModel):
name: str
target_objects: list[str] = Field(default_factory=list)
@@ -1554,6 +1580,25 @@ async def html5_project_access(project_id: str, profile: str | None = None) -> R
)
@app.get("/html5/projects/{project_id}/ai-structure")
async def html5_project_ai_structure(project_id: str) -> Response:
return _html5_response(
_html5_ai_structure_page(project_id=project_id, project_summaries=_project_summaries)
)
@app.post("/html5/projects/{project_id}/ai-structure/run")
async def html5_project_ai_structure_run(project_id: str, request: Request) -> Response:
form = await _html5_form_data(request)
return _html5_response(
_html5_ai_structure_run(
project_id=project_id,
form=form,
prepare=_prepare_ai_structure,
)
)
@app.get("/html5/projects/{project_id}/access/profiles/{profile_name}/plan")
async def html5_project_access_profile_plan(project_id: str, profile_name: str) -> Response:
return _html5_response(
@@ -3318,6 +3363,21 @@ async def runtime_platform() -> dict:
}
@app.post("/ai-structure/prepare", response_model=AiStructurePrepareResponse)
async def prepare_ai_structure_endpoint(request: AiStructurePrepareRequest) -> AiStructurePrepareResponse:
project_id = request.project_id or f"ai-structure-{uuid4()}"
try:
payload = _prepare_ai_structure(
project_id=project_id,
input_path=Path(request.input_path),
output_path=Path(request.output_path),
structure_only=request.structure_only,
)
except FileNotFoundError as error:
raise HTTPException(status_code=404, detail=str(error)) from error
return AiStructurePrepareResponse.model_validate(payload)
@app.get("/storage")
async def storage() -> dict[str, str]:
return {"status": "configured", "path": _storage.root.as_posix()}
@@ -13,8 +13,10 @@
.access-workspace{background:#f4f7fb}.access-layout{display:grid;grid-template-columns:300px minmax(0,1fr)340px;height:calc(100vh - 88px)}.access-main{border-top:0;border-bottom:0;overflow:auto}.access-nav,.access-side{overflow:auto}.tree-item[data-html5-access-profile-selected="true"]{background:#f8fbff;border-left:3px solid var(--brand);padding-left:9px}.access-empty{margin:16px;border:1px dashed #aeb8c6;background:#fff;padding:18px;color:#687385}.access-empty strong,.access-empty span{display:block}.access-empty strong{color:#1f2937}.access-profile,.access-plan,.access-result{border-bottom:1px solid var(--line);background:#fff}.access-summary{display:flex;gap:8px;flex-wrap:wrap;padding:10px 14px;border-bottom:1px solid var(--line);background:#f8fbff;color:#687385;font-size:12px;font-weight:800}.access-role-grid,.access-operations,.access-list{display:grid}.access-role-grid{grid-template-columns:repeat(2,minmax(0,1fr))}.access-card{display:grid;gap:3px;min-width:0;padding:10px 12px;border-bottom:1px solid var(--line)}.access-role-grid .access-card:nth-child(odd){border-right:1px solid var(--line)}.access-card strong,.access-card small{overflow:hidden;text-overflow:ellipsis;white-space:nowrap}.access-card small{color:var(--muted)}.access-plan-head{display:flex;gap:10px;align-items:center;justify-content:space-between;padding:10px 12px;border-bottom:1px solid var(--line);background:#fbfcfe}.access-plan-head form{margin:0}.access-actions{padding:12px}.access-warnings{margin:0;padding:0;list-style:none}.access-warnings li{padding:8px 12px;border-bottom:1px solid var(--line);color:var(--warn);font-weight:800}.access-json{margin:0;max-height:220px;overflow:auto;padding:12px;background:#fbfaf7;border-top:1px solid var(--line);font:12px/1.5 ui-monospace,SFMono-Regular,Consolas,monospace;white-space:pre-wrap}.access-main .primary{background:var(--brand);border-color:var(--brand);color:#fff}
.access-builder{border-bottom:1px solid var(--line);background:#fff}.access-builder-form{display:grid;grid-template-columns:1fr 1fr;gap:10px;padding:12px;border-bottom:1px solid var(--line);background:#fbfcfe}.access-builder-form label{display:grid;gap:5px;font-size:11px;font-weight:900;color:var(--muted);text-transform:uppercase}.access-builder-form input,.access-builder-form textarea{min-width:0;border:1px solid var(--line);background:#fff;padding:0 8px;color:var(--text);font:13px/1.3 system-ui,-apple-system,Segoe UI,sans-serif;text-transform:none}.access-builder-form input{height:32px}.access-builder-form textarea{min-height:68px;padding:8px;resize:vertical}.access-builder-actions{grid-column:1/-1;display:flex;gap:8px;justify-content:flex-end}.access-builder-result{border-top:1px solid var(--line);background:#fff}
.access-card[hx-get]{cursor:pointer}.access-card[hx-get]:hover{background:#f8fbff}.access-user-detail{border-bottom:1px solid var(--line);background:#fff}
.ai-structure-form{display:grid;grid-template-columns:1fr 1fr 220px auto;gap:10px;align-items:end;padding:12px 16px;border-bottom:1px solid var(--line);background:#fbfcfe}.ai-structure-form label{display:grid;gap:5px;font-size:11px;font-weight:900;color:var(--muted);text-transform:uppercase}.ai-structure-form input{height:32px;min-width:0;border:1px solid var(--line);background:#fff;padding:0 8px;color:var(--text);font:13px/1.3 system-ui,-apple-system,Segoe UI,sans-serif;text-transform:none}.ai-structure-result{background:#fff}
@media(max-width:980px){.layout{grid-template-columns:1fr;height:auto}.tree,.inspector{max-height:320px}.editor{min-height:520px}.hero{display:block}.shell{padding:16px}}
@media(max-width:980px){.access-layout{grid-template-columns:1fr;height:auto}.access-nav,.access-side{max-height:360px}.access-role-grid{grid-template-columns:1fr}.access-role-grid .access-card:nth-child(odd){border-right:0}}
@media(max-width:980px){.access-builder-form{grid-template-columns:1fr}.access-builder-actions{justify-content:stretch}.access-builder-actions button{flex:1}}
@media(max-width:980px){.ai-structure-form{grid-template-columns:1fr}}
@media(max-width:980px){.form-designer-body{grid-template-columns:1fr}.form-property-panel{border-left:0;border-top:1px solid var(--line)}.form-field[data-html5-form-width="half"],.form-field[data-html5-form-width="third"]{grid-column:1/-1}.form-editor-row[data-html5-form-element-editor]{grid-template-columns:1fr}.property-row{grid-template-columns:1fr}}
@media(max-width:980px){.setup-layout{grid-template-columns:1fr}.setup-metrics{grid-template-columns:1fr 1fr}.settings-form{grid-template-columns:1fr}}
+65
View File
@@ -1672,6 +1672,71 @@ def test_import_supports_structure_only_indexing(monkeypatch, tmp_path: Path):
assert {"HTTP-сервисы", "Подсистемы", "Последовательности", "Нумераторы документов"}.issubset(common_labels)
def test_ai_structure_prepare_writes_ai_ready_package(tmp_path: Path):
source = tmp_path / "source"
output = tmp_path / "ai-out"
source.mkdir()
(source / "metadata.xml").write_text(
"""
<Configuration>
<Catalog name="Контрагенты" qualifiedName="Справочник.Контрагенты">
<Attribute name="ИНН" qualifiedName="Справочник.Контрагенты.ИНН" />
</Catalog>
<CommonModule name="Интеграция" qualifiedName="ОбщийМодуль.Интеграция" />
</Configuration>
""",
encoding="utf-8",
)
(source / "Интеграция.bsl").write_text(
"Процедура Выполнить() Экспорт\nКонецПроцедуры\n",
encoding="utf-8",
)
client = TestClient(app)
response = client.post(
"/ai-structure/prepare",
json={"project_id": "ai-demo", "input_path": str(source), "output_path": str(output)},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "ready"
assert payload["snapshot"]["nodes"] >= 2
assert (output / "manifest.json").exists()
assert (output / "normalized_project.json").exists()
assert (output / "sir_snapshot.json").exists()
assert "Treat modules/forms/commands as parts" in (output / "ai_context.md").read_text(encoding="utf-8")
page = client.get("/html5/projects/ai-demo/ai-structure")
assert_html5_response_contract(page, 'data-html5-page="ai-structure"', "Структура для ИИ", full_page=True)
html5_run = client.post(
"/html5/projects/ai-demo/ai-structure/run",
data={"project_id": "ai-demo-html5", "input_path": str(source), "output_path": str(tmp_path / "html5-out")},
)
assert_html5_response_contract(html5_run, "ready", "sir_snapshot.json", "normalized_project.json")
def test_ai_structure_prepare_reports_cf_cfe_export_required(tmp_path: Path):
source = tmp_path / "cf-source"
output = tmp_path / "cf-out"
source.mkdir()
(source / "base.cf").write_bytes(b"binary-cf")
(source / "ext.cfe").write_bytes(b"binary-cfe")
client = TestClient(app)
response = client.post(
"/ai-structure/prepare",
json={"project_id": "binary-demo", "input_path": str(source), "output_path": str(output)},
)
assert response.status_code == 200
payload = response.json()
assert payload["status"] == "export_required"
assert len(payload["binary_1c_files"]) == 2
assert "DumpConfigToFiles" in (output / "export_plan.md").read_text(encoding="utf-8")
def test_import_full_replace_replaces_current_normalized_project(tmp_path: Path):
first = tmp_path / "first"
second = tmp_path / "second"