diff --git a/clickhouse-1c/README.md b/clickhouse-1c/README.md index dc5ceb8..24ebc5d 100644 --- a/clickhouse-1c/README.md +++ b/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С или внешних безопасных + выгрузок, если нужна большая бухгалтерская детализация. diff --git a/clickhouse-1c/detections/build_entity_timeline.sql b/clickhouse-1c/detections/build_entity_timeline.sql index 50833fc..82f43f1 100644 --- a/clickhouse-1c/detections/build_entity_timeline.sql +++ b/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 ( diff --git a/clickhouse-1c/detections/insert_detections.sql b/clickhouse-1c/detections/insert_detections.sql index 6883944..8a01eb9 100644 --- a/clickhouse-1c/detections/insert_detections.sql +++ b/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 ( diff --git a/clickhouse-1c/etl/build_business_event_exports.py b/clickhouse-1c/etl/build_business_event_exports.py new file mode 100644 index 0000000..b9a4496 --- /dev/null +++ b/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()) diff --git a/clickhouse-1c/etl/test_build_business_event_exports.py b/clickhouse-1c/etl/test_build_business_event_exports.py new file mode 100644 index 0000000..7b4f1a0 --- /dev/null +++ b/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() diff --git a/clickhouse-1c/ops/run_ingest_cycle.sh b/clickhouse-1c/ops/run_ingest_cycle.sh index 72669b7..1d3c887 100644 --- a/clickhouse-1c/ops/run_ingest_cycle.sh +++ b/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" diff --git a/docs/1C_BUSINESS_EVENT_LAYER_RU.md b/docs/1C_BUSINESS_EVENT_LAYER_RU.md index 73598c0..93e8d0e 100644 --- a/docs/1C_BUSINESS_EVENT_LAYER_RU.md +++ b/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 со всей +бухгалтерской глубиной. ## Что делать дальше