feat(1c): add read-only business event normalizer

clickhouse-1c/README.md

@@ -46,6 +46,8 @@ File 1C + reglog + host telemetry
 - `.env.example` — переменные окружения.
 - `clickhouse/init/*.sql` — схема БД.
 - `etl/load_1c_exports.py` — loader CSV/JSON выгрузок в raw/core таблицы.
+- `etl/build_business_event_exports.py` — read-only normalizer из
+  `documents/postings/audit` в canonical `business_events/document_changes`.
 - `etl/config.example.yml` — пример ETL-конфига.
 - `docs/1C_BUSINESS_EVENT_LAYER_RU.md` — production contract следующего шага:
   canonical business-event слой для документов/проводок/изменений.
@@ -109,6 +111,7 @@ cp etl/config.example.yml etl/config.yml
 python3 -m venv .venv
 . .venv/bin/activate
 pip install -r etl/requirements.txt
+python etl/build_business_event_exports.py --config etl/config.yml
 python etl/load_1c_exports.py --config etl/config.yml
 ```
 
@@ -199,5 +202,7 @@ Read-only API для руководителя:
 - `document_change_events` — изменения документов/реквизитов;
 - ETL уже умеет принимать эти datasets в `landing/business_events` и
   `landing/document_changes`;
-- дальше нужен только read-only extractor, который будет наполнять их из 1С
-  или внешних безопасных выгрузок без записи в базу.
+- built-in normalizer уже умеет собирать этот слой из существующих read-only
+  выгрузок `documents/postings/audit`;
+- дальше нужен только более богатый extractor из 1С или внешних безопасных
+  выгрузок, если нужна большая бухгалтерская детализация.

clickhouse-1c/detections/build_entity_timeline.sql

@@ -17,6 +17,46 @@ FROM (
 ) AS src
 WHERE src.ref_id NOT IN (SELECT ref_id FROM analytics_1c.entity_timeline);
 
+INSERT INTO analytics_1c.entity_timeline
+SELECT *
+FROM (
+    SELECT
+        ts,
+        'document' AS entity_type,
+        document_id AS entity_id,
+        infobase,
+        user AS actor,
+        'business_events' AS source,
+        concat('business:', event_kind, ':', operation_type) AS event_type,
+        if(event_kind = 'posting', 'low', 'medium') AS severity,
+        greatest(10, toUInt32(round(amount))) AS score,
+        concat('business:', event_id) AS ref_id,
+        concat('Business event ', event_kind, ' ', document_type, ' №', document_number, ' amount=', toString(amount)) AS summary
+    FROM analytics_1c.business_events
+    WHERE document_id != ''
+) AS src
+WHERE src.ref_id NOT IN (SELECT ref_id FROM analytics_1c.entity_timeline);
+
+INSERT INTO analytics_1c.entity_timeline
+SELECT *
+FROM (
+    SELECT
+        ts,
+        'counterparty' AS entity_type,
+        company_entity_key AS entity_id,
+        infobase,
+        user AS actor,
+        'business_events' AS source,
+        concat('company:', event_kind) AS event_type,
+        if(amount >= 50000, 'medium', 'low') AS severity,
+        greatest(10, toUInt32(round(amount))) AS score,
+        concat('business-company:', event_id) AS ref_id,
+        concat('Business activity ', counterparty, ' ', operation_type, ' amount=', toString(amount)) AS summary
+    FROM analytics_1c.business_events
+    WHERE company_entity_key != ''
+) AS src
+WHERE src.ref_id NOT IN (SELECT ref_id FROM analytics_1c.entity_timeline);
+
 INSERT INTO analytics_1c.entity_timeline
 SELECT *
 FROM (
@@ -40,6 +80,26 @@ FROM (
 ) AS src
 WHERE src.ref_id NOT IN (SELECT ref_id FROM analytics_1c.entity_timeline);
 
+INSERT INTO analytics_1c.entity_timeline
+SELECT *
+FROM (
+    SELECT
+        ts,
+        if(document_id != '', 'document', 'counterparty') AS entity_type,
+        if(document_id != '', document_id, company_entity_key) AS entity_id,
+        infobase,
+        user AS actor,
+        'document_changes' AS source,
+        concat('change:', change_kind, ':', field_name) AS event_type,
+        if(risk_tag != '', 'high', 'medium') AS severity,
+        if(risk_tag != '', 70, 35) AS score,
+        concat('change:', change_id) AS ref_id,
+        concat('Change ', change_kind, ' field=', field_name, ' risk=', risk_tag) AS summary
+    FROM analytics_1c.document_change_events
+    WHERE (document_id != '' OR company_entity_key != '')
+) AS src
+WHERE src.ref_id NOT IN (SELECT ref_id FROM analytics_1c.entity_timeline);
+
 INSERT INTO analytics_1c.entity_timeline
 SELECT *
 FROM (

clickhouse-1c/detections/insert_detections.sql

@@ -19,6 +19,51 @@ FROM (
 ) AS src
 WHERE src.detection_id NOT IN (SELECT detection_id FROM analytics_1c.detections);
 
+INSERT INTO analytics_1c.detections
+SELECT *
+FROM (
+    SELECT
+        ts,
+        concat('large_manual_adjustment:', infobase, ':', document_id, ':', toString(toUnixTimestamp(ts)), ':', toString(line_no)) AS detection_id,
+        infobase,
+        'large_manual_adjustment' AS rule_id,
+        'Крупная корректировка проводки' AS rule_title,
+        'document' AS entity_type,
+        document_id AS entity_id,
+        'high' AS severity,
+        75 AS score,
+        concat('Крупная корректировка по документу ', document_id, ' amount=', toString(amount), ' счет ', debit_account, ' -> ', credit_account) AS summary,
+        'open' AS status
+    FROM analytics_1c.business_events
+    WHERE event_kind = 'posting'
+      AND amount >= 50000
+      AND (
+          operation_type ILIKE '%adjust%'
+          OR document_type ILIKE '%Коррект%'
+      )
+) AS src
+WHERE src.detection_id NOT IN (SELECT detection_id FROM analytics_1c.detections);
+
+INSERT INTO analytics_1c.detections
+SELECT *
+FROM (
+    SELECT
+        ts,
+        concat('risky_document_change:', infobase, ':', change_id) AS detection_id,
+        infobase,
+        'risky_document_change' AS rule_id,
+        'Рискованное изменение документа' AS rule_title,
+        if(document_id != '', 'document', 'counterparty') AS entity_type,
+        if(document_id != '', document_id, company_entity_key) AS entity_id,
+        'high' AS severity,
+        70 AS score,
+        concat('Изменение ', change_kind, ' field=', field_name, ' risk=', risk_tag) AS summary,
+        'open' AS status
+    FROM analytics_1c.document_change_events
+    WHERE risk_tag != ''
+) AS src
+WHERE src.detection_id NOT IN (SELECT detection_id FROM analytics_1c.detections);
+
 INSERT INTO analytics_1c.detections
 SELECT *
 FROM (

clickhouse-1c/etl/build_business_event_exports.py

@@ -0,0 +1,345 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import re
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from load_1c_exports import Config, iter_rows, load_config, normalize_ts
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="Build canonical business-event exports from read-only 1C file exports")
+    p.add_argument("--config", required=True, help="Path to YAML config")
+    return p.parse_args()
+
+
+@dataclass(frozen=True)
+class DocumentMeta:
+    infobase: str
+    company_entity_key: str
+    organization: str
+    department: str
+    document_id: str
+    document_number: str
+    document_type: str
+    registrar: str
+    user: str
+    counterparty: str
+    operation_type: str
+    amount: float
+
+
+def collapse_ws(value: str) -> str:
+    return re.sub(r"\s+", " ", value).strip()
+
+
+def normalize_base_path(value: str) -> str:
+    text = value.upper().replace("\\", "/")
+    text = re.sub(r"/+", "/", text)
+    text = re.sub(r"[^0-9A-ZА-ЯЁ:/._ -]+", " ", text)
+    return collapse_ws(text)
+
+
+def normalize_infobase_key(value: str) -> str:
+    text = value.upper()
+    text = re.sub(r"(^|\s)20[0-9]{2}($|\s)", " ", text)
+    text = re.sub(r"[^0-9A-ZА-ЯЁ]+", " ", text)
+    return collapse_ws(text)
+
+
+def canonical_company_entity_key(base_id: str = "", base_path: str = "", infobase: str = "") -> str:
+    if base_id:
+        return f"baseid:{base_id.strip()}"
+    if base_path:
+        normalized_path = normalize_base_path(base_path)
+        if normalized_path:
+            return f"basepath:{normalized_path}"
+    normalized_infobase = normalize_infobase_key(infobase)
+    return f"infobase:{normalized_infobase}" if normalized_infobase else ""
+
+
+def source_format(conf: Config, dataset: str) -> str:
+    return conf.formats.get(dataset, conf.formats.get("default", "jsonl"))
+
+
+def landing_root(conf: Config, dataset: str) -> Path:
+    return Path(conf.landing[dataset])
+
+
+def file_ready(path: Path, conf: Config) -> bool:
+    age_seconds = max(0, int((datetime.now(UTC) - datetime.fromtimestamp(path.stat().st_mtime, UTC)).total_seconds()))
+    return age_seconds >= conf.min_file_age_seconds
+
+
+def iter_dataset_files(conf: Config, dataset: str) -> list[Path]:
+    root = landing_root(conf, dataset)
+    if not root.exists():
+        return []
+    return [p for p in sorted(root.iterdir()) if p.is_file() and file_ready(p, conf)]
+
+
+def stable_id(prefix: str, *parts: Any) -> str:
+    payload = "|".join(str(part or "") for part in parts)
+    digest = hashlib.sha1(payload.encode("utf-8"), usedforsecurity=False).hexdigest()[:16]
+    return f"{prefix}:{digest}"
+
+
+def write_jsonl(path: Path, rows: list[dict[str, Any]]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if not rows:
+        path.unlink(missing_ok=True)
+        return
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    payload = "\n".join(json.dumps(row, ensure_ascii=False) for row in rows) + "\n"
+    tmp.write_text(payload, encoding="utf-8")
+    tmp.replace(path)
+
+
+def load_company_index(conf: Config) -> dict[str, str]:
+    latest: dict[str, tuple[datetime, str]] = {}
+    for path in iter_dataset_files(conf, "companies"):
+        for row in iter_rows(path, source_format(conf, "companies")):
+            infobase = str(row.get("infobase", "")).strip()
+            if not infobase:
+                continue
+            entity_key = canonical_company_entity_key(
+                base_id=str(row.get("base_id", "")).strip(),
+                base_path=str(row.get("base_path", "")).strip(),
+                infobase=infobase,
+            )
+            ts = normalize_ts(row.get("ts"))
+            current = latest.get(infobase)
+            if current is None or ts >= current[0]:
+                latest[infobase] = (ts, entity_key)
+    return {infobase: entity_key for infobase, (_ts, entity_key) in latest.items()}
+
+
+def derive_company_entity_key(row: dict[str, Any], company_index: dict[str, str]) -> str:
+    explicit = str(row.get("company_entity_key", "")).strip()
+    if explicit:
+        return explicit
+    infobase = str(row.get("infobase", "")).strip()
+    return company_index.get(infobase, "") or canonical_company_entity_key(
+        base_id=str(row.get("base_id", "")).strip(),
+        base_path=str(row.get("base_path", "")).strip(),
+        infobase=infobase,
+    )
+
+
+def build_document_index(conf: Config, company_index: dict[str, str]) -> tuple[dict[tuple[str, str], DocumentMeta], dict[tuple[str, str], DocumentMeta]]:
+    rows: list[dict[str, Any]] = []
+    for path in iter_dataset_files(conf, "documents"):
+        rows.extend(iter_rows(path, source_format(conf, "documents")))
+    return build_document_index_from_rows(rows, company_index)
+
+
+def build_document_index_from_rows(rows: list[dict[str, Any]], company_index: dict[str, str]) -> tuple[dict[tuple[str, str], DocumentMeta], dict[tuple[str, str], DocumentMeta]]:
+    by_id: dict[tuple[str, str], DocumentMeta] = {}
+    by_number: dict[tuple[str, str], DocumentMeta] = {}
+    for row in rows:
+        infobase = str(row.get("infobase", "")).strip()
+        document_id = str(row.get("doc_id", row.get("document_id", ""))).strip()
+        document_number = str(row.get("doc_number", row.get("document_number", ""))).strip()
+        if not infobase:
+            continue
+        meta = DocumentMeta(
+            infobase=infobase,
+            company_entity_key=derive_company_entity_key(row, company_index),
+            organization=str(row.get("organization", "")).strip(),
+            department=str(row.get("department", "")).strip(),
+            document_id=document_id,
+            document_number=document_number,
+            document_type=str(row.get("doc_type", row.get("document_type", ""))).strip(),
+            registrar=document_id or document_number,
+            user=str(row.get("author", row.get("user", ""))).strip(),
+            counterparty=str(row.get("counterparty", "")).strip(),
+            operation_type=str(row.get("operation_type", "")).strip(),
+            amount=float(row.get("amount", 0) or 0),
+        )
+        if document_id:
+            by_id[(infobase, document_id)] = meta
+        if document_number:
+            by_number[(infobase, document_number)] = meta
+    return by_id, by_number
+
+
+def lookup_document_meta(
+    row: dict[str, Any],
+    by_id: dict[tuple[str, str], DocumentMeta],
+    by_number: dict[tuple[str, str], DocumentMeta],
+) -> DocumentMeta | None:
+    infobase = str(row.get("infobase", "")).strip()
+    registrar = str(row.get("registrar", row.get("document_id", row.get("doc_id", "")))).strip()
+    document_number = str(row.get("document_number", row.get("doc_number", ""))).strip()
+    if infobase and registrar and (infobase, registrar) in by_id:
+        return by_id[(infobase, registrar)]
+    if infobase and document_number and (infobase, document_number) in by_number:
+        return by_number[(infobase, document_number)]
+    return None
+
+
+def build_document_events(rows: list[dict[str, Any]], source_file: str, company_index: dict[str, str]) -> list[dict[str, Any]]:
+    events: list[dict[str, Any]] = []
+    for row in rows:
+        infobase = str(row.get("infobase", "")).strip()
+        document_id = str(row.get("doc_id", row.get("document_id", ""))).strip()
+        document_number = str(row.get("doc_number", row.get("document_number", ""))).strip()
+        document_type = str(row.get("doc_type", row.get("document_type", ""))).strip()
+        ts = normalize_ts(row.get("ts") or row.get("posted_at") or row.get("created_at"))
+        company_entity_key = derive_company_entity_key(row, company_index)
+        events.append(
+            {
+                "ts": ts.isoformat(),
+                "event_id": stable_id("document_snapshot", source_file, infobase, document_id, document_number, ts.isoformat()),
+                "infobase": infobase,
+                "company_entity_key": company_entity_key,
+                "organization": str(row.get("organization", "")).strip(),
+                "department": str(row.get("department", "")).strip(),
+                "document_id": document_id,
+                "document_number": document_number,
+                "document_type": document_type,
+                "registrar": document_id or document_number,
+                "operation_type": str(row.get("operation_type", "")).strip(),
+                "event_kind": "document_snapshot",
+                "user": str(row.get("author", row.get("user", ""))).strip(),
+                "counterparty": str(row.get("counterparty", "")).strip(),
+                "counterparty_inn": str(row.get("counterparty_inn", "")).strip(),
+                "debit_account": "",
+                "credit_account": "",
+                "amount": float(row.get("amount", 0) or 0),
+                "currency": str(row.get("currency", "RUB")).strip() or "RUB",
+                "line_no": 0,
+                "evidence_ref": f"document:{document_id or document_number}",
+            }
+        )
+    return events
+
+
+def build_posting_events(
+    rows: list[dict[str, Any]],
+    source_file: str,
+    company_index: dict[str, str],
+    by_id: dict[tuple[str, str], DocumentMeta],
+    by_number: dict[tuple[str, str], DocumentMeta],
+) -> list[dict[str, Any]]:
+    events: list[dict[str, Any]] = []
+    for idx, row in enumerate(rows, start=1):
+        meta = lookup_document_meta(row, by_id, by_number)
+        infobase = str(row.get("infobase", "")).strip()
+        registrar = str(row.get("registrar", "")).strip()
+        ts = normalize_ts(row.get("ts"))
+        company_entity_key = meta.company_entity_key if meta else derive_company_entity_key(row, company_index)
+        line_no = int(row.get("line_no", idx) or idx)
+        events.append(
+            {
+                "ts": ts.isoformat(),
+                "event_id": stable_id("posting", source_file, infobase, registrar, line_no, ts.isoformat()),
+                "infobase": infobase,
+                "company_entity_key": company_entity_key,
+                "organization": meta.organization if meta else str(row.get("organization", "")).strip(),
+                "department": meta.department if meta else str(row.get("department", "")).strip(),
+                "document_id": meta.document_id if meta else registrar,
+                "document_number": meta.document_number if meta else str(row.get("document_number", "")).strip(),
+                "document_type": meta.document_type if meta else str(row.get("document_type", "")).strip(),
+                "registrar": registrar,
+                "operation_type": str(row.get("operation_type", meta.operation_type if meta else "")).strip(),
+                "event_kind": "posting",
+                "user": meta.user if meta else str(row.get("user", row.get("author", ""))).strip(),
+                "counterparty": meta.counterparty if meta else str(row.get("counterparty", "")).strip(),
+                "counterparty_inn": str(row.get("counterparty_inn", "")).strip(),
+                "debit_account": str(row.get("account_dt", row.get("debit_account", ""))).strip(),
+                "credit_account": str(row.get("account_ct", row.get("credit_account", ""))).strip(),
+                "amount": float(row.get("amount", 0) or 0),
+                "currency": str(row.get("currency", "RUB")).strip() or "RUB",
+                "line_no": line_no,
+                "evidence_ref": f"posting:{registrar}:{line_no}",
+            }
+        )
+    return events
+
+
+def build_document_changes(
+    rows: list[dict[str, Any]],
+    source_file: str,
+    company_index: dict[str, str],
+    by_id: dict[tuple[str, str], DocumentMeta],
+    by_number: dict[tuple[str, str], DocumentMeta],
+) -> list[dict[str, Any]]:
+    changes: list[dict[str, Any]] = []
+    for row in rows:
+        infobase = str(row.get("infobase", "")).strip()
+        object_type = str(row.get("object_type", "")).strip()
+        object_id = str(row.get("object_id", "")).strip()
+        ts = normalize_ts(row.get("ts"))
+        meta = lookup_document_meta(
+            {
+                "infobase": infobase,
+                "registrar": row.get("document_id") or (object_id if object_type == "document" else ""),
+                "document_number": row.get("document_number", ""),
+            },
+            by_id,
+            by_number,
+        )
+        company_entity_key = meta.company_entity_key if meta else derive_company_entity_key(row, company_index)
+        document_id = meta.document_id if meta else (object_id if object_type == "document" else str(row.get("document_id", "")).strip())
+        changes.append(
+            {
+                "ts": ts.isoformat(),
+                "change_id": stable_id("change", source_file, infobase, object_type, object_id, row.get("action", ""), ts.isoformat()),
+                "infobase": infobase,
+                "company_entity_key": company_entity_key,
+                "organization": meta.organization if meta else str(row.get("organization", "")).strip(),
+                "document_id": document_id,
+                "document_number": meta.document_number if meta else str(row.get("document_number", "")).strip(),
+                "document_type": meta.document_type if meta else str(row.get("document_type", "")).strip(),
+                "change_kind": str(row.get("change_kind", row.get("action", object_type))).strip(),
+                "field_name": str(row.get("field_name", object_type)).strip(),
+                "user": str(row.get("user", row.get("author", ""))).strip(),
+                "before_value": str(row.get("before_value", row.get("before_hash", ""))).strip(),
+                "after_value": str(row.get("after_value", row.get("after_hash", ""))).strip(),
+                "risk_tag": str(row.get("risk_tag", "")).strip(),
+                "evidence_ref": f"audit:{object_type}:{object_id}",
+            }
+        )
+    return changes
+
+
+def output_path(conf: Config, dataset: str, source_path: Path, prefix: str) -> Path:
+    return landing_root(conf, dataset) / f"{prefix}-{source_path.stem}.jsonl"
+
+
+def main() -> int:
+    args = parse_args()
+    conf = load_config(args.config)
+    company_index = load_company_index(conf)
+    by_id, by_number = build_document_index(conf, company_index)
+
+    for path in iter_dataset_files(conf, "documents"):
+        rows = iter_rows(path, source_format(conf, "documents"))
+        out = output_path(conf, "business_events", path, "business-events-documents")
+        write_jsonl(out, build_document_events(rows, path.name, company_index))
+        print(f"built business_events from documents: {path.name} rows={len(rows)}")
+
+    for path in iter_dataset_files(conf, "postings"):
+        rows = iter_rows(path, source_format(conf, "postings"))
+        out = output_path(conf, "business_events", path, "business-events-postings")
+        write_jsonl(out, build_posting_events(rows, path.name, company_index, by_id, by_number))
+        print(f"built business_events from postings: {path.name} rows={len(rows)}")
+
+    for path in iter_dataset_files(conf, "audit"):
+        rows = iter_rows(path, source_format(conf, "audit"))
+        out = output_path(conf, "document_changes", path, "document-changes-audit")
+        write_jsonl(out, build_document_changes(rows, path.name, company_index, by_id, by_number))
+        print(f"built document_changes from audit: {path.name} rows={len(rows)}")
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

clickhouse-1c/etl/test_build_business_event_exports.py

@@ -0,0 +1,104 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import sys
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+sys.modules.setdefault("clickhouse_connect", SimpleNamespace(get_client=lambda **_: None))
+
+import build_business_event_exports as builder
+
+
+class BuildBusinessEventExportsTests(unittest.TestCase):
+    def test_canonical_company_entity_key(self) -> None:
+        self.assertEqual(builder.canonical_company_entity_key(base_id="ABC"), "baseid:ABC")
+        self.assertTrue(builder.canonical_company_entity_key(base_path=r"c:\1c\Base 1").startswith("basepath:"))
+        self.assertEqual(builder.canonical_company_entity_key(infobase="ФЕЛИЦТ ГРУПП 2026"), "infobase:ФЕЛИЦТ ГРУПП")
+
+    def test_build_document_and_posting_events(self) -> None:
+        company_index = {"ТРАНСГАЗ 2026": "baseid:tgz"}
+        docs = [
+            {
+                "ts": "2026-05-22T10:00:00Z",
+                "infobase": "ТРАНСГАЗ 2026",
+                "organization": "Трансгаз",
+                "department": "Продажи",
+                "doc_type": "Реализация",
+                "doc_id": "DOC-1",
+                "doc_number": "0001",
+                "author": "USER1",
+                "counterparty": "ООО Альфа",
+                "operation_type": "sale",
+                "amount": "125000.00",
+            }
+        ]
+        by_id, by_number = builder.build_document_index_from_rows(docs, company_index)
+        doc_events = builder.build_document_events(docs, "docs.jsonl", company_index)
+        posting_events = builder.build_posting_events(
+            [
+                {
+                    "ts": "2026-05-22T10:01:00Z",
+                    "infobase": "ТРАНСГАЗ 2026",
+                    "registrar": "DOC-1",
+                    "operation_type": "sale",
+                    "account_dt": "62.01",
+                    "account_ct": "90.01",
+                    "amount": "125000.00",
+                }
+            ],
+            "postings.jsonl",
+            company_index,
+            by_id,
+            by_number,
+        )
+        self.assertEqual(doc_events[0]["company_entity_key"], "baseid:tgz")
+        self.assertEqual(doc_events[0]["event_kind"], "document_snapshot")
+        self.assertEqual(posting_events[0]["document_type"], "Реализация")
+        self.assertEqual(posting_events[0]["debit_account"], "62.01")
+        self.assertEqual(posting_events[0]["line_no"], 1)
+
+    def test_build_document_changes(self) -> None:
+        company_index = {"ФЕЛИЦТ ГРУПП 2026": "infobase:ФЕЛИЦТ ГРУПП"}
+        docs = [
+            {
+                "ts": "2026-05-22T10:00:00Z",
+                "infobase": "ФЕЛИЦТ ГРУПП 2026",
+                "organization": "Фелицт",
+                "doc_type": "Корректировка",
+                "doc_id": "DOC-3",
+                "doc_number": "0003",
+                "author": "USER4",
+                "counterparty": "ООО Бета",
+            }
+        ]
+        by_id, by_number = builder.build_document_index_from_rows(docs, company_index)
+        changes = builder.build_document_changes(
+            [
+                {
+                    "ts": "2026-05-22T10:02:00Z",
+                    "infobase": "ФЕЛИЦТ ГРУПП 2026",
+                    "user": "USER4",
+                    "object_type": "document",
+                    "object_id": "DOC-3",
+                    "action": "repost",
+                    "before_hash": "abc",
+                    "after_hash": "def",
+                    "risk_tag": "repost",
+                }
+            ],
+            "audit.jsonl",
+            company_index,
+            by_id,
+            by_number,
+        )
+        self.assertEqual(changes[0]["document_id"], "DOC-3")
+        self.assertEqual(changes[0]["document_type"], "Корректировка")
+        self.assertEqual(changes[0]["change_kind"], "repost")
+        self.assertEqual(changes[0]["risk_tag"], "repost")
+
+
+if __name__ == "__main__":
+    unittest.main()

clickhouse-1c/ops/run_ingest_cycle.sh

@@ -40,6 +40,7 @@ fi
 # shellcheck disable=SC1090
 . "${ENV_FILE}"
 
+"${VENV}/bin/python" "${ROOT}/etl/build_business_event_exports.py" --config "${CONFIG}"
 "${VENV}/bin/python" "${ROOT}/etl/load_1c_exports.py" --config "${CONFIG}"
 "${VENV}/bin/python" "${ROOT}/etl/load_company_registry_xlsx.py" --config "${CONFIG}" --landing "${ROOT}/landing/registry"
 

docs/1C_BUSINESS_EVENT_LAYER_RU.md

@@ -179,8 +179,21 @@
   - `landing/business_events`
   - `landing/document_changes`
   - dataset mapping в `etl/load_1c_exports.py`
+- built-in normalizer:
+  - `etl/build_business_event_exports.py`
+  - собирает canonical events из существующих read-only выгрузок
+    `documents/postings/audit`
+- ingest wiring:
+  - normalizer запускается перед `load_1c_exports.py`
+- timeline/detection wiring:
+  - `business_events` и `document_change_events` уже входят в
+    `entity_timeline`
+  - на этом слое уже есть первые detections:
+    - крупная корректировка проводки
+    - рискованное изменение документа
 
-Это именно scaffold, а не обещание, что live extractor уже существует.
+Это уже рабочий v1 normalizer, но ещё не конечный extractor со всей
+бухгалтерской глубиной.
 
 ## Что делать дальше