diff --git a/ansible/deploy_file_1c_analytics.yml b/ansible/deploy_file_1c_analytics.yml index 3fa1999..4320b9f 100644 --- a/ansible/deploy_file_1c_analytics.yml +++ b/ansible/deploy_file_1c_analytics.yml @@ -19,6 +19,15 @@ aw_file_1c_grafana_port: 3300 aw_file_1c_company_api_host: "{{ ansible_host | default('127.0.0.1') }}" aw_file_1c_company_api_port: 8710 + aw_file_1c_manager_brief_state_dir: /opt/activitywatch/clickhouse-1c/state/manager-brief + aw_file_1c_manager_brief_model: gpt-5.3-codex + aw_file_1c_manager_brief_codex_user: codex + aw_file_1c_manager_brief_codex_bin: codex + aw_file_1c_manager_brief_workdir: /home/codex/infra-admin + aw_file_1c_manager_brief_top_limit: 5 + aw_file_1c_manager_brief_freshness_hours: 8 + aw_file_1c_manager_brief_timeout_sec: 300 + aw_file_1c_manager_brief_run_after_ingest: false aw_file_1c_windows_upload_pubkey_path: /tmp/awops_ed25519.pub tasks: @@ -106,6 +115,7 @@ - check_ingest_freshness.sh - run_company_intelligence_api.sh - run_company_intelligence_refresh.sh + - run_manager_brief.sh - run_ingest_cycle.sh - name: Создать .env для file-1C analytics @@ -126,6 +136,15 @@ CLICKHOUSE_HOST=clickhouse AW_1C_COMPANY_API_HOST={{ aw_file_1c_company_api_host }} AW_1C_COMPANY_API_PORT={{ aw_file_1c_company_api_port }} + AW_1C_MANAGER_BRIEF_STATE_DIR={{ aw_file_1c_manager_brief_state_dir }} + AW_1C_MANAGER_BRIEF_MODEL={{ aw_file_1c_manager_brief_model }} + AW_1C_MANAGER_BRIEF_CODEX_USER={{ aw_file_1c_manager_brief_codex_user }} + AW_1C_MANAGER_BRIEF_CODEX_BIN={{ aw_file_1c_manager_brief_codex_bin }} + AW_1C_MANAGER_BRIEF_WORKDIR={{ aw_file_1c_manager_brief_workdir }} + AW_1C_MANAGER_BRIEF_TOP_LIMIT={{ aw_file_1c_manager_brief_top_limit }} + AW_1C_MANAGER_BRIEF_FRESHNESS_HOURS={{ aw_file_1c_manager_brief_freshness_hours }} + AW_1C_MANAGER_BRIEF_TIMEOUT_SEC={{ aw_file_1c_manager_brief_timeout_sec }} + AW_1C_MANAGER_BRIEF_RUN_AFTER_INGEST={{ 1 if aw_file_1c_manager_brief_run_after_ingest else 0 }} - name: Создать etl/config.yml для file-1C analytics ansible.builtin.copy: @@ -254,6 +273,24 @@ mode: "0644" notify: Перезагрузить systemd + - name: Установить systemd unit aw-1c-manager-brief.service + ansible.builtin.copy: + src: "{{ aw_file_1c_repo_root }}/clickhouse-1c/ops/aw-1c-manager-brief.service" + dest: /etc/systemd/system/aw-1c-manager-brief.service + owner: root + group: root + mode: "0644" + notify: Перезагрузить systemd + + - name: Установить systemd unit aw-1c-manager-brief.timer + ansible.builtin.copy: + src: "{{ aw_file_1c_repo_root }}/clickhouse-1c/ops/aw-1c-manager-brief.timer" + dest: /etc/systemd/system/aw-1c-manager-brief.timer + owner: root + group: root + mode: "0644" + notify: Перезагрузить systemd + - name: Разрешить Windows upload key для igor ansible.builtin.lineinfile: path: /home/igor/.ssh/authorized_keys @@ -284,6 +321,13 @@ state: started daemon_reload: true + - name: Включить и запустить aw-1c-manager-brief.timer + ansible.builtin.systemd: + name: aw-1c-manager-brief.timer + enabled: true + state: started + daemon_reload: true + - name: Проверить доступность ClickHouse ping ansible.builtin.uri: url: "http://127.0.0.1:{{ aw_file_1c_clickhouse_port }}/ping" diff --git a/clickhouse-1c/.env.example b/clickhouse-1c/.env.example index c842680..a3d17ac 100644 --- a/clickhouse-1c/.env.example +++ b/clickhouse-1c/.env.example @@ -13,3 +13,12 @@ AW_1C_COMPANY_API_PORT=8710 AW_1C_COMPANY_LOOKBACK_DAYS=30 AW_1C_COMPANY_MIN_DAYS=1 AW_1C_COMPANY_HORIZONS=7,30 +AW_1C_MANAGER_BRIEF_STATE_DIR=/opt/activitywatch/clickhouse-1c/state/manager-brief +AW_1C_MANAGER_BRIEF_MODEL=gpt-5.3-codex +AW_1C_MANAGER_BRIEF_CODEX_USER=codex +AW_1C_MANAGER_BRIEF_CODEX_BIN=codex +AW_1C_MANAGER_BRIEF_WORKDIR=/home/codex/infra-admin +AW_1C_MANAGER_BRIEF_TOP_LIMIT=5 +AW_1C_MANAGER_BRIEF_FRESHNESS_HOURS=8 +AW_1C_MANAGER_BRIEF_TIMEOUT_SEC=300 +AW_1C_MANAGER_BRIEF_RUN_AFTER_INGEST=0 diff --git a/clickhouse-1c/README.md b/clickhouse-1c/README.md index 4c32353..05e6b73 100644 --- a/clickhouse-1c/README.md +++ b/clickhouse-1c/README.md @@ -58,6 +58,9 @@ File 1C + reglog + host telemetry - `ai/INVESTIGATOR_API.md` — контракт AI Investigator поверх ClickHouse/cases. - `ai/refresh_company_intelligence.py` — materialization forecast/signals по `counterparty`. - `ai/company_intelligence_api.py` — read-only API для AI/аналитики по компаниям. +- `ai/generate_manager_brief.py` — executive brief для руководителя поверх live company intelligence. +- `ai/manager_brief_prompt.md` — prompt для локального `codex exec`. +- `ai/manager_brief_schema.json` — строгая schema structured-brief ответа. ## Когда использовать именно этот контур @@ -124,9 +127,31 @@ python ai/refresh_company_intelligence.py --host localhost --port 8123 --user de python ai/company_intelligence_api.py --host 127.0.0.1 --port 8710 ``` -8. В Grafana строить dashboards из `grafana/dashboard-catalog.md` и +8. Сформировать executive brief: + +```bash +python ai/generate_manager_brief.py --host localhost --port 8123 --user default --password change-me --database analytics_1c +``` + +9. В Grafana строить dashboards из `grafana/dashboard-catalog.md` и `grafana/query-pack.sql`. +## Manager brief + +Этот слой делает не raw LLM-чат, а промышленный pipeline: + +- строит компактный context из `v_company_portfolio_overview`; +- вызывает локальный `codex exec` на `10.10.10.2`; +- валидирует structured output по JSON schema; +- при сбое `codex` отдаёт deterministic fallback, чтобы контур не пустел; +- пишет `latest.json` и `latest.md` в `state/manager-brief/`. + +Read-only API для руководителя: + +- `GET /api/1/analytics-1c/manager/brief/latest` +- `GET /api/1/analytics-1c/manager/brief/latest.md` +- `GET /api/1/analytics-1c/manager/brief/history` + ## Ожидаемые источники данных - выгрузки 1С по документам; diff --git a/clickhouse-1c/ai/company_intelligence_api.py b/clickhouse-1c/ai/company_intelligence_api.py index 6c58804..0edb673 100644 --- a/clickhouse-1c/ai/company_intelligence_api.py +++ b/clickhouse-1c/ai/company_intelligence_api.py @@ -2,14 +2,17 @@ from __future__ import annotations import argparse +import json import os from datetime import UTC, datetime from decimal import Decimal +from pathlib import Path from typing import Any import clickhouse_connect import uvicorn from fastapi import FastAPI, HTTPException, Query +from fastapi.responses import PlainTextResponse def parse_args() -> argparse.Namespace: @@ -48,6 +51,21 @@ def ch_client(): ) +def manager_brief_state_dir() -> Path: + root = Path(os.getenv("AW_1C_ROOT", "/opt/activitywatch/clickhouse-1c")) + configured = os.getenv("AW_1C_MANAGER_BRIEF_STATE_DIR") + if configured: + return Path(configured) + return root / "state" / "manager-brief" + + +def load_latest_manager_brief() -> dict[str, Any]: + latest_path = manager_brief_state_dir() / "latest.json" + if not latest_path.exists(): + raise HTTPException(status_code=404, detail="manager brief not generated yet") + return json.loads(latest_path.read_text(encoding="utf-8")) + + app = FastAPI(title="AW-rus 1C Company Intelligence API", version="1.0.0") @@ -229,6 +247,39 @@ def company_timeline( return {"items": rows, "count": len(rows)} +@app.get("/api/1/analytics-1c/manager/brief/latest") +def manager_brief_latest() -> dict[str, Any]: + return load_latest_manager_brief() + + +@app.get("/api/1/analytics-1c/manager/brief/latest.md", response_class=PlainTextResponse) +def manager_brief_latest_markdown() -> str: + latest_md = manager_brief_state_dir() / "latest.md" + if not latest_md.exists(): + raise HTTPException(status_code=404, detail="manager brief markdown not generated yet") + return latest_md.read_text(encoding="utf-8") + + +@app.get("/api/1/analytics-1c/manager/brief/history") +def manager_brief_history(limit: int = Query(default=20, ge=1, le=200)) -> dict[str, Any]: + history_dir = manager_brief_state_dir() / "history" + if not history_dir.exists(): + return {"items": [], "count": 0} + items: list[dict[str, Any]] = [] + for path in sorted(history_dir.glob("*.json"), reverse=True)[:limit]: + payload = json.loads(path.read_text(encoding="utf-8")) + items.append( + { + "generated_at": payload.get("generated_at"), + "render_mode": payload.get("render_mode"), + "model": payload.get("model"), + "headline": payload.get("brief", {}).get("headline", ""), + "path": path.name, + } + ) + return {"items": items, "count": len(items)} + + if __name__ == "__main__": args = parse_args() uvicorn.run(app, host=args.host, port=args.port) diff --git a/clickhouse-1c/ai/generate_manager_brief.py b/clickhouse-1c/ai/generate_manager_brief.py new file mode 100644 index 0000000..75031db --- /dev/null +++ b/clickhouse-1c/ai/generate_manager_brief.py @@ -0,0 +1,512 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import argparse +import json +import os +import shlex +import subprocess +import tempfile +from datetime import UTC, datetime +from decimal import Decimal +from pathlib import Path +from typing import Any + +ROOT = Path(__file__).resolve().parents[1] +PROMPT_PATH = ROOT / "ai" / "manager_brief_prompt.md" +SCHEMA_PATH = ROOT / "ai" / "manager_brief_schema.json" + + +def parse_args() -> argparse.Namespace: + p = argparse.ArgumentParser(description="Generate executive manager brief for analytics_1c") + p.add_argument("--host", default=os.getenv("CLICKHOUSE_HOST", "localhost")) + p.add_argument("--port", type=int, default=int(os.getenv("CLICKHOUSE_PORT", "8123"))) + p.add_argument("--user", default=os.getenv("CLICKHOUSE_USER", "default")) + p.add_argument("--password", default=os.getenv("CLICKHOUSE_PASSWORD", "")) + p.add_argument("--database", default=os.getenv("CLICKHOUSE_DB", "analytics_1c")) + p.add_argument( + "--state-dir", + default=os.getenv("AW_1C_MANAGER_BRIEF_STATE_DIR", str(ROOT / "state" / "manager-brief")), + ) + p.add_argument( + "--codex-user", + default=os.getenv("AW_1C_MANAGER_BRIEF_CODEX_USER", "codex"), + ) + p.add_argument( + "--codex-bin", + default=os.getenv("AW_1C_MANAGER_BRIEF_CODEX_BIN", "codex"), + ) + p.add_argument( + "--workdir", + default=os.getenv("AW_1C_MANAGER_BRIEF_WORKDIR", "/home/codex/infra-admin"), + ) + p.add_argument( + "--model", + default=os.getenv("AW_1C_MANAGER_BRIEF_MODEL", "gpt-5.3-codex"), + ) + p.add_argument( + "--top-limit", + type=int, + default=int(os.getenv("AW_1C_MANAGER_BRIEF_TOP_LIMIT", "5")), + ) + p.add_argument( + "--freshness-hours", + type=int, + default=int(os.getenv("AW_1C_MANAGER_BRIEF_FRESHNESS_HOURS", "8")), + ) + p.add_argument( + "--timeout-sec", + type=int, + default=int(os.getenv("AW_1C_MANAGER_BRIEF_TIMEOUT_SEC", "300")), + ) + return p.parse_args() + + +def to_plain(value: Any) -> Any: + if isinstance(value, Decimal): + return float(value) + if isinstance(value, datetime): + return value.isoformat() + return value + + +def rows_to_dict(result) -> list[dict[str, Any]]: + return [ + {name: to_plain(value) for name, value in zip(result.column_names, row)} + for row in result.result_rows + ] + + +def ch_client(args: argparse.Namespace): + import clickhouse_connect + + return clickhouse_connect.get_client( + host=args.host, + port=args.port, + username=args.user, + password=args.password, + database=args.database, + ) + + +def load_text(path: Path) -> str: + return path.read_text(encoding="utf-8") + + +def write_text(path: Path, content: str) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(content, encoding="utf-8") + + +def write_json(path: Path, payload: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") + + +def q(value: str) -> str: + return "'" + value.replace("'", "''") + "'" + + +def build_context(client, top_limit: int, freshness_hours: int) -> dict[str, Any]: + now = datetime.now(UTC) + + portfolio_summary = rows_to_dict( + client.query( + """ + SELECT + count() AS companies_total, + countIf(signal_severity = 'critical') AS critical_total, + countIf(signal_severity = 'high') AS high_total, + countIf(signal_severity = 'medium') AS medium_total, + countIf(signal_severity = 'low') AS low_total, + countIf(signal_severity = 'none') AS none_total, + countIf(registry_match_mode = 'direct') AS direct_total, + countIf(registry_match_mode = 'alias') AS alias_total, + countIf(registry_match_mode = 'manual') AS manual_total, + countIf(registry_match_mode = 'none') AS unmatched_total, + countIf(days_since_last_activity >= 7) AS stale_7d_total, + countIf(days_since_last_activity >= 14) AS stale_14d_total, + countIf(current_status = 'busy' OR active_locks > 0 OR temp_db_present > 0) AS busy_total, + round(sum(amount_30d), 2) AS activity_30d_total, + round(sum(amount_forecast_30d), 2) AS activity_forecast_30d_total, + sum(open_cases_total) AS open_cases_total, + sum(detections_total) AS detections_total + FROM analytics_1c.v_company_portfolio_overview + """ + ) + )[0] + + freshness = rows_to_dict( + client.query( + """ + SELECT + (SELECT max(ts) FROM analytics_1c.documents) AS documents_ts, + (SELECT max(ts) FROM analytics_1c.companies) AS companies_ts, + (SELECT max(ts) FROM analytics_1c.reglog_events) AS reglog_ts, + (SELECT max(ts) FROM analytics_1c.audit_events) AS audit_ts, + (SELECT max(ts) FROM analytics_1c.host_events) AS host_ts, + (SELECT max(generated_at) FROM analytics_1c.company_forecasts) AS forecasts_ts, + (SELECT max(generated_at) FROM analytics_1c.company_health_signals) AS signals_ts + """ + ) + )[0] + freshness_items: list[dict[str, Any]] = [] + for source, ts in freshness.items(): + lag_hours = None + stale = True + if isinstance(ts, str): + parsed = datetime.fromisoformat(ts) + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=UTC) + lag_hours = round((now - parsed).total_seconds() / 3600, 2) + stale = lag_hours > freshness_hours + ts = parsed.isoformat() + freshness_items.append( + { + "source": source, + "latest_ts": ts, + "lag_hours": lag_hours, + "stale": stale, + } + ) + + top_risks = rows_to_dict( + client.query( + f""" + SELECT + infobase, + counterparty, + normalized_counterparty, + registry_match_mode, + registry_assignee_name, + signal_severity, + signal_score, + top_signal, + current_status, + active_locks, + open_cases_total, + detections_total, + days_since_last_activity, + round(amount_30d, 2) AS amount_30d, + round(amount_forecast_30d, 2) AS amount_forecast_30d + FROM analytics_1c.v_company_portfolio_overview + ORDER BY signal_score DESC, open_cases_total DESC, detections_total DESC, amount_30d DESC, counterparty + LIMIT {int(top_limit)} + """ + ) + ) + + top_forecasts = rows_to_dict( + client.query( + f""" + SELECT + infobase, + counterparty, + normalized_counterparty, + registry_match_mode, + signal_severity, + signal_score, + round(amount_30d, 2) AS amount_30d, + round(amount_forecast_30d, 2) AS amount_forecast_30d, + round(docs_forecast_30d, 2) AS docs_forecast_30d, + amount_forecast_confidence, + top_signal + FROM analytics_1c.v_company_portfolio_overview + ORDER BY amount_forecast_30d DESC, signal_score DESC, amount_30d DESC, counterparty + LIMIT {int(top_limit)} + """ + ) + ) + + watchlist = rows_to_dict( + client.query( + f""" + SELECT + s.infobase, + s.counterparty, + p.normalized_counterparty, + p.registry_match_mode, + s.signal_type, + s.severity, + s.score, + s.summary, + p.days_since_last_activity, + round(p.amount_30d, 2) AS amount_30d, + round(p.amount_forecast_30d, 2) AS amount_forecast_30d + FROM analytics_1c.v_company_health_current AS s + LEFT JOIN analytics_1c.v_company_portfolio_overview AS p + ON p.infobase = s.infobase AND p.counterparty = s.counterparty + WHERE s.signal_type IN ('inactive_company', 'amount_drop', 'docs_stopped') + ORDER BY s.score DESC, p.days_since_last_activity DESC, p.amount_30d DESC + LIMIT {int(top_limit)} + """ + ) + ) + + busy_bases = rows_to_dict( + client.query( + f""" + SELECT + infobase, + counterparty, + normalized_counterparty, + current_status, + active_locks, + temp_db_present, + scheduler_touched, + current_activity_score, + signal_severity, + signal_score + FROM analytics_1c.v_company_portfolio_overview + WHERE current_status = 'busy' OR active_locks > 0 OR temp_db_present > 0 + ORDER BY active_locks DESC, temp_db_present DESC, current_activity_score DESC, counterparty + LIMIT {int(top_limit)} + """ + ) + ) + + recent_cases = rows_to_dict( + client.query( + f""" + SELECT + c.opened_at, + c.infobase, + c.entity_id AS counterparty, + c.title, + c.severity, + c.status + FROM analytics_1c.cases AS c + WHERE c.entity_type = 'counterparty' AND c.status != 'closed' + ORDER BY c.opened_at DESC + LIMIT {int(top_limit)} + """ + ) + ) + + return { + "generated_at": now.isoformat(), + "freshness_threshold_hours": freshness_hours, + "portfolio_summary": portfolio_summary, + "freshness": freshness_items, + "top_risks": top_risks, + "top_forecasts": top_forecasts, + "watchlist": watchlist, + "busy_bases": busy_bases, + "recent_cases": recent_cases, + } + + +def render_deterministic_payload(context: dict[str, Any]) -> dict[str, Any]: + summary = context["portfolio_summary"] + freshness = context["freshness"] + stale_sources = [item["source"] for item in freshness if item["stale"]] + top_risks = context["top_risks"][:5] + top_forecasts = context["top_forecasts"][:5] + watchlist = context["watchlist"][:5] + + headline = ( + f"Портфель {summary['companies_total']} компаний: критичных {summary['critical_total']}, " + f"high {summary['high_total']}, stale 14д {summary['stale_14d_total']}." + ) + summary_lines = [ + f"Покрытие реестра полное: direct {summary['direct_total']}, alias {summary['alias_total']}, manual {summary['manual_total']}, unmatched {summary['unmatched_total']}.", + f"Суммарная активность за 30 дней {summary['activity_30d_total']}, прогнозная активность на 30 дней {summary['activity_forecast_30d_total']}.", + f"Открытых кейсов по компаниям {summary['open_cases_total']}, активных detections {summary['detections_total']}.", + ] + if stale_sources: + summary_lines.append(f"Есть просрочка по источникам: {', '.join(stale_sources)}.") + else: + summary_lines.append("Свежесть источников укладывается в заданный порог.") + + risk_items = [ + { + "company": item["counterparty"], + "severity": item["signal_severity"], + "reason": item["top_signal"] or "Повышенный signal score без детализации top_signal.", + "recommended_action": ( + "Проверить открытые кейсы, detections и фактическую занятость файловой базы." + if item["open_cases_total"] or item["detections_total"] or item["active_locks"] + else "Проверить последние события по компании и причину роста operational severity." + ), + } + for item in top_risks + ] + + forecast_items = [ + { + "company": item["counterparty"], + "forecast_30d": str(item["amount_forecast_30d"]), + "interpretation": ( + f"Текущая активность 30д {item['amount_30d']}, match {item['registry_match_mode']}, severity {item['signal_severity']}." + ), + } + for item in top_forecasts + ] + + actions = [ + "Разобрать компании с открытыми кейсами и максимальным signal score в первую очередь.", + "Проверить watchlist по inactivity/amount_drop/docs_stopped и подтвердить, это бизнес-пауза или operational сбой.", + "Отдельно пройти по manual-match компаниям перед управленческими выводами из реестра.", + ] + if watchlist: + actions[1] = ( + f"Проверить watchlist: {', '.join(item['counterparty'] for item in watchlist[:3])}." + ) + + caveats = [ + "Показатель amount здесь трактуется как activity score, а не как деньги или выручка.", + "Severity operational-driven: high/critical отражают кейсы, detections и занятость базы, а не автоматически финансовый риск.", + ] + if summary["manual_total"] > 0: + caveats.append("Компании с registry_match_mode=manual требуют осторожности при юридической интерпретации реестра.") + + return { + "headline": headline, + "summary": summary_lines[:6], + "top_risks": risk_items[:5], + "top_forecasts": forecast_items[:5], + "actions": actions[:5], + "caveats": caveats[:4], + } + + +def render_markdown(payload: dict[str, Any], generated_at: str) -> str: + lines = [ + f"# Executive Brief 1C", + "", + f"_Сформировано: {generated_at}_", + "", + f"## Заголовок", + payload["headline"], + "", + "## Кратко", + ] + for item in payload["summary"]: + lines.append(f"- {item}") + lines.extend(["", "## Компании риска"]) + for idx, item in enumerate(payload["top_risks"], start=1): + lines.append( + f"{idx}. {item['company']} [{item['severity']}] — {item['reason']} Действие: {item['recommended_action']}" + ) + lines.extend(["", "## Прогноз по активности 30д"]) + for idx, item in enumerate(payload["top_forecasts"], start=1): + lines.append( + f"{idx}. {item['company']} — прогноз {item['forecast_30d']}. {item['interpretation']}" + ) + lines.extend(["", "## Рекомендуемые действия"]) + for item in payload["actions"]: + lines.append(f"- {item}") + lines.extend(["", "## Ограничения"]) + for item in payload["caveats"]: + lines.append(f"- {item}") + lines.append("") + return "\n".join(lines) + + +def run_codex(prompt: str, args: argparse.Namespace) -> tuple[int, str, str]: + output_file = Path(tempfile.mkstemp(prefix="aw-1c-manager-brief-", suffix=".json")[1]) + os.chmod(output_file, 0o666) + cmd_inner = ( + f"cd {shlex.quote(args.workdir)} && " + f"{shlex.quote(args.codex_bin)} exec --ephemeral --skip-git-repo-check " + f"--model {shlex.quote(args.model)} " + f"-C {shlex.quote(args.workdir)} " + f"-s read-only " + f"--color never " + f"--output-schema {shlex.quote(str(SCHEMA_PATH))} " + f"-o {shlex.quote(str(output_file))} -" + ) + if os.geteuid() == 0 and args.codex_user: + cmd = ["sudo", "-u", args.codex_user, "-H", "bash", "-lc", cmd_inner] + else: + cmd = ["bash", "-lc", cmd_inner] + + try: + result = subprocess.run( + cmd, + input=prompt, + text=True, + capture_output=True, + timeout=args.timeout_sec, + check=False, + ) + reply = output_file.read_text(encoding="utf-8").strip() if output_file.exists() else "" + stdout_stderr = (result.stdout or "") + ("\n" + result.stderr if result.stderr else "") + return result.returncode, stdout_stderr.strip(), reply + finally: + try: + output_file.unlink() + except FileNotFoundError: + pass + + +def build_prompt(context: dict[str, Any]) -> str: + template = load_text(PROMPT_PATH) + return template.replace( + "{{CONTEXT_JSON}}", + json.dumps(context, ensure_ascii=False, indent=2), + ) + + +def save_artifacts( + state_dir: Path, + artifact: dict[str, Any], + markdown: str, +) -> None: + timestamp = datetime.fromisoformat(artifact["generated_at"]).strftime("%Y%m%dT%H%M%SZ") + history_dir = state_dir / "history" + history_dir.mkdir(parents=True, exist_ok=True) + latest_json = state_dir / "latest.json" + latest_md = state_dir / "latest.md" + history_json = history_dir / f"{timestamp}.json" + history_md = history_dir / f"{timestamp}.md" + write_json(latest_json, artifact) + write_text(latest_md, markdown) + write_json(history_json, artifact) + write_text(history_md, markdown) + + +def main() -> int: + args = parse_args() + state_dir = Path(args.state_dir) + state_dir.mkdir(parents=True, exist_ok=True) + + client = ch_client(args) + context = build_context(client, top_limit=args.top_limit, freshness_hours=args.freshness_hours) + prompt = build_prompt(context) + + codex_rc = None + codex_output = "" + render_mode = "deterministic" + payload: dict[str, Any] + + try: + codex_rc, codex_output, codex_reply = run_codex(prompt, args) + payload = json.loads(codex_reply) if codex_reply else {} + if not payload: + raise ValueError("empty codex payload") + render_mode = "codex" + except Exception as exc: # noqa: BLE001 + payload = render_deterministic_payload(context) + codex_output = f"{codex_output}\nFALLBACK: {exc}".strip() + render_mode = "deterministic" + + generated_at = datetime.now(UTC).replace(microsecond=0).isoformat() + markdown = render_markdown(payload, generated_at) + artifact = { + "generated_at": generated_at, + "render_mode": render_mode, + "model": args.model, + "codex_rc": codex_rc, + "context": context, + "brief": payload, + "markdown": markdown, + "codex_output_excerpt": codex_output[-4000:] if codex_output else "", + } + save_artifacts(state_dir, artifact, markdown) + print(json.dumps({"status": "ok", "render_mode": render_mode, "state_dir": str(state_dir)}, ensure_ascii=False)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/clickhouse-1c/ai/manager_brief_prompt.md b/clickhouse-1c/ai/manager_brief_prompt.md new file mode 100644 index 0000000..9389af6 --- /dev/null +++ b/clickhouse-1c/ai/manager_brief_prompt.md @@ -0,0 +1,18 @@ +Ты готовишь краткий executive brief для руководителя по портфелю компаний в файловой 1С. + +Правила: +- Пиши только по фактам из переданного JSON-контекста. +- Не придумывай данные и не достраивай причины, которых нет в контексте. +- Пиши по-русски, коротко, без технического мусора. +- Не упоминай ИИ, Codex, модель, prompt, JSON, ClickHouse, API. +- Если в данных `amount` означает activity score, называй это "активность", а не "выручка" или "деньги". +- Если почти все компании в high/critical, явно скажи, что severity сейчас operational-driven и не равна финансовому кризису. +- Для `registry_match_mode=manual` не делай сильных выводов о юридическом соответствии реестру. +- Приоритет: риски, прогноз, что проверить руководителю в первую очередь. + +Верни JSON строго по schema. + +Контекст: + +{{CONTEXT_JSON}} + diff --git a/clickhouse-1c/ai/manager_brief_schema.json b/clickhouse-1c/ai/manager_brief_schema.json new file mode 100644 index 0000000..fe0b00e --- /dev/null +++ b/clickhouse-1c/ai/manager_brief_schema.json @@ -0,0 +1,115 @@ +{ + "type": "object", + "additionalProperties": false, + "required": [ + "headline", + "summary", + "top_risks", + "top_forecasts", + "actions", + "caveats" + ], + "properties": { + "headline": { + "type": "string", + "minLength": 1, + "maxLength": 180 + }, + "summary": { + "type": "array", + "minItems": 3, + "maxItems": 6, + "items": { + "type": "string", + "minLength": 1, + "maxLength": 260 + } + }, + "top_risks": { + "type": "array", + "maxItems": 5, + "items": { + "type": "object", + "additionalProperties": false, + "required": [ + "company", + "severity", + "reason", + "recommended_action" + ], + "properties": { + "company": { + "type": "string", + "minLength": 1, + "maxLength": 160 + }, + "severity": { + "type": "string", + "minLength": 1, + "maxLength": 32 + }, + "reason": { + "type": "string", + "minLength": 1, + "maxLength": 260 + }, + "recommended_action": { + "type": "string", + "minLength": 1, + "maxLength": 260 + } + } + } + }, + "top_forecasts": { + "type": "array", + "maxItems": 5, + "items": { + "type": "object", + "additionalProperties": false, + "required": [ + "company", + "forecast_30d", + "interpretation" + ], + "properties": { + "company": { + "type": "string", + "minLength": 1, + "maxLength": 160 + }, + "forecast_30d": { + "type": "string", + "minLength": 1, + "maxLength": 64 + }, + "interpretation": { + "type": "string", + "minLength": 1, + "maxLength": 260 + } + } + } + }, + "actions": { + "type": "array", + "minItems": 2, + "maxItems": 5, + "items": { + "type": "string", + "minLength": 1, + "maxLength": 260 + } + }, + "caveats": { + "type": "array", + "minItems": 1, + "maxItems": 4, + "items": { + "type": "string", + "minLength": 1, + "maxLength": 260 + } + } + } +} diff --git a/clickhouse-1c/ai/test_manager_brief.py b/clickhouse-1c/ai/test_manager_brief.py new file mode 100644 index 0000000..60505c3 --- /dev/null +++ b/clickhouse-1c/ai/test_manager_brief.py @@ -0,0 +1,79 @@ +#!/usr/bin/env python3 +from __future__ import annotations + +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import generate_manager_brief as gmb + + +class ManagerBriefTests(unittest.TestCase): + def setUp(self) -> None: + self.context = { + "portfolio_summary": { + "companies_total": 41, + "critical_total": 10, + "high_total": 20, + "medium_total": 5, + "low_total": 3, + "none_total": 3, + "direct_total": 33, + "alias_total": 5, + "manual_total": 3, + "unmatched_total": 0, + "stale_7d_total": 4, + "stale_14d_total": 2, + "busy_total": 7, + "activity_30d_total": 12345.0, + "activity_forecast_30d_total": 54321.0, + "open_cases_total": 12, + "detections_total": 20, + }, + "freshness": [ + {"source": "documents_ts", "stale": False}, + {"source": "signals_ts", "stale": True}, + ], + "top_risks": [ + { + "counterparty": "ФЕЛИЦТ ГРУПП 2026", + "signal_severity": "critical", + "top_signal": "Есть открытые кейсы", + "open_cases_total": 2, + "detections_total": 5, + "active_locks": 3, + "amount_30d": 100.0, + "amount_forecast_30d": 300.0, + } + ], + "top_forecasts": [ + { + "counterparty": "АВКО 2026", + "amount_30d": 1000.0, + "amount_forecast_30d": 3000.0, + "registry_match_mode": "direct", + "signal_severity": "high", + } + ], + "watchlist": [{"counterparty": "АВКО 2026"}], + } + + def test_deterministic_payload_shape(self) -> None: + payload = gmb.render_deterministic_payload(self.context) + self.assertIn("headline", payload) + self.assertGreaterEqual(len(payload["summary"]), 3) + self.assertEqual(payload["top_risks"][0]["company"], "ФЕЛИЦТ ГРУПП 2026") + self.assertIn("manual", " ".join(payload["caveats"]).lower()) + + def test_markdown_render(self) -> None: + payload = gmb.render_deterministic_payload(self.context) + md = gmb.render_markdown(payload, "2026-05-22T12:00:00+00:00") + self.assertIn("# Executive Brief 1C", md) + self.assertIn("## Компании риска", md) + self.assertIn("ФЕЛИЦТ ГРУПП 2026", md) + + +if __name__ == "__main__": + unittest.main() diff --git a/clickhouse-1c/ops/aw-1c-manager-brief.service b/clickhouse-1c/ops/aw-1c-manager-brief.service new file mode 100644 index 0000000..cf2d35e --- /dev/null +++ b/clickhouse-1c/ops/aw-1c-manager-brief.service @@ -0,0 +1,12 @@ +[Unit] +Description=AW-rus 1C Manager Executive Brief +After=network-online.target aw-1c-company-api.service +Wants=network-online.target + +[Service] +Type=oneshot +Environment=AW_1C_ROOT=/opt/activitywatch/clickhouse-1c +ExecStart=/opt/activitywatch/clickhouse-1c/ops/run_manager_brief.sh + +[Install] +WantedBy=multi-user.target diff --git a/clickhouse-1c/ops/aw-1c-manager-brief.timer b/clickhouse-1c/ops/aw-1c-manager-brief.timer new file mode 100644 index 0000000..0807616 --- /dev/null +++ b/clickhouse-1c/ops/aw-1c-manager-brief.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Run AW-rus 1C manager brief every 6 hours + +[Timer] +OnBootSec=30min +OnUnitActiveSec=6h +Unit=aw-1c-manager-brief.service +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/clickhouse-1c/ops/bootstrap_runtime.sh b/clickhouse-1c/ops/bootstrap_runtime.sh index 9ce7568..1123a07 100644 --- a/clickhouse-1c/ops/bootstrap_runtime.sh +++ b/clickhouse-1c/ops/bootstrap_runtime.sh @@ -17,7 +17,8 @@ mkdir -p \ "${ROOT}/archive/registry" \ "${ROOT}/archive/reglog" \ "${ROOT}/archive/audit" \ - "${ROOT}/archive/host" + "${ROOT}/archive/host" \ + "${ROOT}/state/manager-brief/history" if [[ ! -f "${ROOT}/etl/config.yml" ]]; then cp "${ROOT}/etl/config.example.yml" "${ROOT}/etl/config.yml" diff --git a/clickhouse-1c/ops/run_ingest_cycle.sh b/clickhouse-1c/ops/run_ingest_cycle.sh index 655d80b..a376845 100644 --- a/clickhouse-1c/ops/run_ingest_cycle.sh +++ b/clickhouse-1c/ops/run_ingest_cycle.sh @@ -65,3 +65,9 @@ docker exec -i "${CH_CONTAINER}" clickhouse-client \ --password "${CLICKHOUSE_PASSWORD}" \ --database "${CLICKHOUSE_DB}" \ < "${ROOT}/detections/open_cases_from_detections.sql" + +if [[ "${AW_1C_MANAGER_BRIEF_RUN_AFTER_INGEST:-0}" == "1" ]]; then + if ! "${ROOT}/ops/run_manager_brief.sh"; then + echo "warning: manager brief refresh failed after ingest" >&2 + fi +fi diff --git a/clickhouse-1c/ops/run_manager_brief.sh b/clickhouse-1c/ops/run_manager_brief.sh new file mode 100644 index 0000000..34ebc9a --- /dev/null +++ b/clickhouse-1c/ops/run_manager_brief.sh @@ -0,0 +1,35 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT="${AW_1C_ROOT:-/opt/activitywatch/clickhouse-1c}" +ENV_FILE="${ROOT}/.env" +VENV="${ROOT}/.venv" + +if [[ ! -f "${ENV_FILE}" ]]; then + echo "missing env file: ${ENV_FILE}" >&2 + exit 1 +fi + +if [[ ! -x "${VENV}/bin/python" ]]; then + echo "missing venv python: ${VENV}/bin/python" >&2 + exit 1 +fi + +# shellcheck disable=SC1090 +set -a +. "${ENV_FILE}" +set +a + +CH_RUNTIME_HOST="${AW_1C_CLICKHOUSE_RUNTIME_HOST:-${CLICKHOUSE_HOST}}" +if [[ "${CH_RUNTIME_HOST}" == "clickhouse" ]]; then + CH_RUNTIME_HOST="127.0.0.1" +fi + +export CLICKHOUSE_HOST="${CH_RUNTIME_HOST}" + +exec "${VENV}/bin/python" "${ROOT}/ai/generate_manager_brief.py" \ + --host "${CH_RUNTIME_HOST}" \ + --port "${CLICKHOUSE_PORT}" \ + --user "${CLICKHOUSE_USER}" \ + --password "${CLICKHOUSE_PASSWORD}" \ + --database "${CLICKHOUSE_DB}" diff --git a/docs/1C_COMPANY_INTELLIGENCE_RU.md b/docs/1C_COMPANY_INTELLIGENCE_RU.md index 4c5bcca..1626df8 100644 --- a/docs/1C_COMPANY_INTELLIGENCE_RU.md +++ b/docs/1C_COMPANY_INTELLIGENCE_RU.md @@ -87,6 +87,34 @@ Endpoints: - `GET /api/1/analytics-1c/companies/{counterparty}/summary` - `GET /api/1/analytics-1c/companies/{counterparty}/forecast` - `GET /api/1/analytics-1c/companies/{counterparty}/timeline` +- `GET /api/1/analytics-1c/manager/brief/latest` +- `GET /api/1/analytics-1c/manager/brief/latest.md` +- `GET /api/1/analytics-1c/manager/brief/history` + +### Executive brief для руководителя + +Файлы: + +- `clickhouse-1c/ai/generate_manager_brief.py` +- `clickhouse-1c/ai/manager_brief_prompt.md` +- `clickhouse-1c/ai/manager_brief_schema.json` +- `clickhouse-1c/ops/run_manager_brief.sh` +- `clickhouse-1c/ops/aw-1c-manager-brief.service` +- `clickhouse-1c/ops/aw-1c-manager-brief.timer` + +Что делает: + +- собирает live context по портфелю компаний из `ClickHouse`; +- вызывает локальный `codex exec` на `10.10.10.2` от пользователя `codex`; +- требует structured JSON по schema, а не свободный текст; +- рендерит итог в `latest.json` и `latest.md`; +- при сбое `codex` даёт deterministic fallback, чтобы контур не оставался пустым. + +Важно: + +- `tmux` не является production-зависимостью; +- интерактивная сессия `codex` может быть открыта, но pipeline работает через обычный `codex exec`; +- service не трогает `1С`, работает только на уже выгруженных read-only данных. ### Ops @@ -133,6 +161,12 @@ clickhouse-client --queries-file clickhouse/init/04_company_intelligence.sql ./ops/run_company_intelligence_api.sh ``` +### 5. Сформировать manager brief + +```bash +./ops/run_manager_brief.sh +``` + По умолчанию: - host: `127.0.0.1` @@ -151,6 +185,15 @@ clickhouse-client --queries-file clickhouse/init/04_company_intelligence.sql - `AW_1C_COMPANY_LOOKBACK_DAYS` - `AW_1C_COMPANY_MIN_DAYS` - `AW_1C_COMPANY_HORIZONS` +- `AW_1C_MANAGER_BRIEF_STATE_DIR` +- `AW_1C_MANAGER_BRIEF_MODEL` +- `AW_1C_MANAGER_BRIEF_CODEX_USER` +- `AW_1C_MANAGER_BRIEF_CODEX_BIN` +- `AW_1C_MANAGER_BRIEF_WORKDIR` +- `AW_1C_MANAGER_BRIEF_TOP_LIMIT` +- `AW_1C_MANAGER_BRIEF_FRESHNESS_HOURS` +- `AW_1C_MANAGER_BRIEF_TIMEOUT_SEC` +- `AW_1C_MANAGER_BRIEF_RUN_AFTER_INGEST` ## Что уже есть в live payload