feat(1c): ingest company registry workbook

clickhouse-1c/ai/company_intelligence_api.py

@@ -60,6 +60,7 @@ def health() -> dict[str, Any]:
             SELECT
                 countIf(counterparty != '') AS documents_with_counterparty,
                 (SELECT count() FROM analytics_1c.companies) AS companies_total,
+                (SELECT count() FROM analytics_1c.company_registry) AS registry_rows_total,
                 (SELECT count() FROM analytics_1c.company_forecasts) AS forecasts_total,
                 (SELECT count() FROM analytics_1c.company_health_signals) AS health_signals_total
             FROM analytics_1c.documents
@@ -85,6 +86,12 @@ def companies_overview(
         organization,
         counterparty,
         company_name,
+        registry_assignee_name,
+        registry_status,
+        registry_share_text,
+        registry_key_contour,
+        registry_inn,
+        registry_kpp,
         owner_user,
         base_path,
         current_status,

clickhouse-1c/clickhouse/init/01_raw_tables.sql

@@ -25,6 +25,16 @@ CREATE TABLE IF NOT EXISTS analytics_1c.raw_1c_companies
 ENGINE = MergeTree
 ORDER BY (ingested_at, source_file);
 
+CREATE TABLE IF NOT EXISTS analytics_1c.raw_1c_company_registry
+(
+    ingested_at DateTime DEFAULT now(),
+    source_file String,
+    source_sheet String,
+    payload String
+)
+ENGINE = MergeTree
+ORDER BY (ingested_at, source_file, source_sheet);
+
 CREATE TABLE IF NOT EXISTS analytics_1c.raw_reglog
 (
     ingested_at DateTime DEFAULT now(),

clickhouse-1c/clickhouse/init/02_core_tables.sql

@@ -53,6 +53,23 @@ CREATE TABLE IF NOT EXISTS analytics_1c.companies
 ENGINE = MergeTree
 ORDER BY (infobase, ts);
 
+CREATE TABLE IF NOT EXISTS analytics_1c.company_registry
+(
+    ts DateTime,
+    source_file String,
+    source_sheet LowCardinality(String),
+    company_name String,
+    company_key String,
+    assignee_name String,
+    registry_status LowCardinality(String),
+    share_text String,
+    key_contour UInt8,
+    inn String,
+    kpp String
+)
+ENGINE = MergeTree
+ORDER BY (company_key, ts, source_sheet);
+
 CREATE TABLE IF NOT EXISTS analytics_1c.reglog_events
 (
     ts DateTime,

clickhouse-1c/clickhouse/init/04_company_intelligence.sql

@@ -101,6 +101,20 @@ SELECT
 FROM analytics_1c.companies
 GROUP BY infobase;
 
+CREATE OR REPLACE VIEW analytics_1c.v_company_registry_current AS
+SELECT
+    company_key,
+    argMax(company_name, ts) AS company_name,
+    argMax(assignee_name, ts) AS assignee_name,
+    argMax(registry_status, ts) AS registry_status,
+    argMax(share_text, ts) AS share_text,
+    argMax(key_contour, ts) AS key_contour,
+    argMax(inn, ts) AS inn,
+    argMax(kpp, ts) AS kpp,
+    max(ts) AS last_registry_snapshot_at
+FROM analytics_1c.company_registry
+GROUP BY company_key;
+
 CREATE OR REPLACE VIEW analytics_1c.v_company_portfolio_overview AS
 WITH
 base AS
@@ -137,6 +151,11 @@ company_state AS
     SELECT *
     FROM analytics_1c.v_companies_current
 ),
+registry_state AS
+(
+    SELECT *
+    FROM analytics_1c.v_company_registry_current
+),
 signals AS
 (
     SELECT
@@ -196,6 +215,12 @@ SELECT
     if(company_state.organization != '', company_state.organization, base.organization) AS organization,
     base.counterparty AS counterparty,
     if(company_state.company_name != '', company_state.company_name, base.counterparty) AS company_name,
+    ifNull(registry_state.assignee_name, '') AS registry_assignee_name,
+    ifNull(registry_state.registry_status, '') AS registry_status,
+    ifNull(registry_state.share_text, '') AS registry_share_text,
+    ifNull(registry_state.key_contour, 0) AS registry_key_contour,
+    ifNull(registry_state.inn, '') AS registry_inn,
+    ifNull(registry_state.kpp, '') AS registry_kpp,
     ifNull(company_state.owner_user, '') AS owner_user,
     ifNull(company_state.base_id, '') AS base_id,
     ifNull(company_state.base_path, '') AS base_path,
@@ -229,6 +254,7 @@ SELECT
     ifNull(signals.top_signal, '') AS top_signal
 FROM base
 LEFT JOIN company_state ON company_state.infobase = base.infobase
+LEFT JOIN registry_state ON registry_state.company_key = trimBoth(replaceRegexpAll(replaceRegexpAll(replaceRegexpAll(upperUTF8(base.counterparty), '(^|\\s)20[0-9]{2}($|\\s)', ' '), '[^0-9A-ZА-ЯЁ]+', ' '), '\\s+', ' '))
 LEFT JOIN d7 ON d7.infobase = base.infobase AND d7.counterparty = base.counterparty
 LEFT JOIN d30 ON d30.infobase = base.infobase AND d30.counterparty = base.counterparty
 LEFT JOIN signals ON signals.infobase = base.infobase AND signals.counterparty = base.counterparty

clickhouse-1c/etl/load_company_registry_xlsx.py

@@ -0,0 +1,191 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import argparse
+import json
+import re
+import shutil
+from dataclasses import dataclass
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+import clickhouse_connect
+import yaml
+from openpyxl import load_workbook
+
+
+@dataclass
+class Config:
+    clickhouse: dict[str, Any]
+    archive_dir: str | None
+    delete_after_load: bool
+    min_file_age_seconds: int
+
+
+def parse_args() -> argparse.Namespace:
+    p = argparse.ArgumentParser(description="Load company registry xlsx into ClickHouse")
+    p.add_argument("--config", required=True)
+    p.add_argument("--landing", required=True)
+    return p.parse_args()
+
+
+def load_config(path: str) -> Config:
+    raw = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
+    return Config(
+        clickhouse=raw["clickhouse"],
+        archive_dir=raw.get("archive_dir"),
+        delete_after_load=bool(raw.get("delete_after_load", False)),
+        min_file_age_seconds=int(raw.get("min_file_age_seconds", 180)),
+    )
+
+
+def ch_client(conf: Config):
+    return clickhouse_connect.get_client(
+        host=conf.clickhouse["host"],
+        port=conf.clickhouse.get("port", 8123),
+        username=conf.clickhouse.get("username", "default"),
+        password=conf.clickhouse.get("password", ""),
+        database=conf.clickhouse.get("database", "analytics_1c"),
+    )
+
+
+def normalize_company_key(value: str) -> str:
+    text = (value or "").upper().replace("Ё", "Е")
+    text = re.sub(r"(^|\s)20\d{2}($|\s)", " ", text)
+    text = re.sub(r"[^0-9A-ZА-Я]+", " ", text)
+    text = re.sub(r"\s+", " ", text)
+    return text.strip()
+
+
+def as_text(value: Any) -> str:
+    if value is None:
+        return ""
+    return str(value).strip()
+
+
+def archive_or_delete(conf: Config, path: Path) -> None:
+    if conf.archive_dir:
+        archive_root = Path(conf.archive_dir) / "registry"
+        archive_root.mkdir(parents=True, exist_ok=True)
+        shutil.move(str(path), archive_root / path.name)
+        return
+    if conf.delete_after_load:
+        path.unlink(missing_ok=True)
+
+
+def parse_registry(path: Path) -> list[dict[str, Any]]:
+    wb = load_workbook(path, read_only=True, data_only=True)
+    now = datetime.now(UTC).replace(tzinfo=None, microsecond=0)
+
+    tax_map: dict[str, tuple[str, str]] = {}
+    if "Лист2" in wb.sheetnames:
+        ws = wb["Лист2"]
+        for row in ws.iter_rows(min_row=3, values_only=True):
+            company_name = as_text(row[1] if len(row) > 1 else "")
+            if not company_name:
+                continue
+            tax_map[normalize_company_key(company_name)] = (
+                as_text(row[2] if len(row) > 2 else ""),
+                as_text(row[3] if len(row) > 3 else ""),
+            )
+
+    rows: list[dict[str, Any]] = []
+    if "ОСНОВНОЙ" not in wb.sheetnames:
+        return rows
+
+    ws = wb["ОСНОВНОЙ"]
+    top_headers = [as_text(v) for v in next(ws.iter_rows(min_row=1, max_row=1, values_only=True))]
+    manager_headers = [as_text(v) for v in next(ws.iter_rows(min_row=2, max_row=2, values_only=True))]
+
+    col_specs: list[dict[str, Any]] = []
+    current_manager: dict[str, Any] | None = None
+    for idx in range(1, len(manager_headers)):
+        manager = manager_headers[idx]
+        top = top_headers[idx] if idx < len(top_headers) else ""
+        if manager:
+            current_manager = {
+                "col": idx,
+                "assignee_name": manager,
+                "meta_col": None,
+                "meta_label": "",
+                "registry_status": "active",
+            }
+            col_specs.append(current_manager)
+            continue
+        if top and "исключ" in top.lower():
+            current_manager = {
+                "col": idx,
+                "assignee_name": "",
+                "meta_col": None,
+                "meta_label": "",
+                "registry_status": "excluded",
+            }
+            col_specs.append(current_manager)
+            continue
+        if current_manager is not None:
+            current_manager["meta_col"] = idx
+            current_manager["meta_label"] = top or "meta"
+            current_manager = None
+
+    for row in ws.iter_rows(min_row=3, values_only=True):
+        for spec in col_specs:
+            company_name = as_text(row[spec["col"]] if spec["col"] < len(row) else "")
+            if not company_name:
+                continue
+            meta_value = ""
+            if spec.get("meta_col") is not None:
+                meta_value = as_text(row[spec["meta_col"]] if spec["meta_col"] < len(row) else "")
+            company_key = normalize_company_key(company_name)
+            inn, kpp = tax_map.get(company_key, ("", ""))
+            rows.append(
+                {
+                    "ts": now,
+                    "source_file": path.name,
+                    "source_sheet": "ОСНОВНОЙ",
+                    "company_name": company_name,
+                    "company_key": company_key,
+                    "assignee_name": spec["assignee_name"],
+                    "registry_status": spec["registry_status"],
+                    "share_text": meta_value if "ключ" not in spec.get("meta_label", "").lower() else "",
+                    "key_contour": 1 if meta_value.upper() == "ЕСТЬ" and "ключ" in spec.get("meta_label", "").lower() else 0,
+                    "inn": inn,
+                    "kpp": kpp,
+                }
+            )
+
+    return rows
+
+
+def main() -> int:
+    args = parse_args()
+    conf = load_config(args.config)
+    client = ch_client(conf)
+    landing = Path(args.landing)
+    if not landing.exists():
+        return 0
+
+    for path in sorted(p for p in landing.iterdir() if p.is_file() and p.suffix.lower() == ".xlsx"):
+        age_seconds = max(0, int((datetime.now(UTC) - datetime.fromtimestamp(path.stat().st_mtime, UTC)).total_seconds()))
+        if age_seconds < conf.min_file_age_seconds:
+            print(f"skip registry: {path.name} age={age_seconds}s < min_file_age_seconds={conf.min_file_age_seconds}")
+            continue
+        rows = parse_registry(path)
+        if rows:
+            client.insert(
+                "analytics_1c.raw_1c_company_registry",
+                [[row["source_file"], row["source_sheet"], json.dumps({k: (v.isoformat() if isinstance(v, datetime) else v) for k, v in row.items()}, ensure_ascii=False)] for row in rows],
+                column_names=["source_file", "source_sheet", "payload"],
+            )
+            client.insert(
+                "analytics_1c.company_registry",
+                [[row["ts"], row["source_file"], row["source_sheet"], row["company_name"], row["company_key"], row["assignee_name"], row["registry_status"], row["share_text"], row["key_contour"], row["inn"], row["kpp"]] for row in rows],
+                column_names=["ts", "source_file", "source_sheet", "company_name", "company_key", "assignee_name", "registry_status", "share_text", "key_contour", "inn", "kpp"],
+            )
+            print(f"loaded registry: {path.name} rows={len(rows)}")
+        archive_or_delete(conf, path)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

clickhouse-1c/etl/requirements.txt

@@ -1,3 +1,4 @@
 clickhouse-connect>=0.7.16
 PyYAML>=6.0.2
 python-dateutil>=2.9.0
+openpyxl>=3.1.5

clickhouse-1c/grafana/provisioning/dashboards/files/1c-company-intelligence.json

@@ -352,7 +352,7 @@
           "format": 1,
           "pluginVersion": "11.2.2",
           "queryType": "table",
-          "rawSql": "SELECT infobase, company_name, owner_user, current_status, round(db_size_bytes / 1048576, 2) AS db_size_mb, amount_30d, docs_30d, amount_forecast_30d, signal_severity, signal_score, top_signal, last_seen_at FROM analytics_1c.v_company_portfolio_overview ORDER BY signal_score DESC, amount_30d DESC LIMIT 20",
+          "rawSql": "SELECT infobase, company_name, registry_assignee_name, registry_inn, registry_kpp, owner_user, current_status, round(db_size_bytes / 1048576, 2) AS db_size_mb, amount_30d, docs_30d, amount_forecast_30d, signal_severity, signal_score, top_signal, last_seen_at FROM analytics_1c.v_company_portfolio_overview ORDER BY signal_score DESC, amount_30d DESC LIMIT 20",
           "refId": "A"
         }
       ],
@@ -475,6 +475,6 @@
   "timezone": "browser",
   "title": "1C File - Company Intelligence",
   "uid": "1c-file-companies",
-  "version": 3,
+  "version": 4,
   "weekStart": ""
 }

clickhouse-1c/ops/bootstrap_runtime.sh

@@ -7,12 +7,14 @@ mkdir -p \
   "${ROOT}/landing/documents" \
   "${ROOT}/landing/postings" \
   "${ROOT}/landing/companies" \
+  "${ROOT}/landing/registry" \
   "${ROOT}/landing/reglog" \
   "${ROOT}/landing/audit" \
   "${ROOT}/landing/host" \
   "${ROOT}/archive/documents" \
   "${ROOT}/archive/postings" \
   "${ROOT}/archive/companies" \
+  "${ROOT}/archive/registry" \
   "${ROOT}/archive/reglog" \
   "${ROOT}/archive/audit" \
   "${ROOT}/archive/host"

clickhouse-1c/ops/run_ingest_cycle.sh

@@ -38,6 +38,7 @@ fi
 . "${ENV_FILE}"
 
 "${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"
 
 docker exec -i "${CH_CONTAINER}" clickhouse-client \
   --user "${CLICKHOUSE_USER}" \