1034 lines
46 KiB
Python
1034 lines
46 KiB
Python
#!/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 severity_rank(value: str | None) -> int:
|
||
return {
|
||
"none": 0,
|
||
"low": 1,
|
||
"medium": 2,
|
||
"high": 3,
|
||
"critical": 4,
|
||
}.get((value or "none").lower(), 0)
|
||
|
||
|
||
def rank_top_change(change: dict[str, Any]) -> dict[str, Any]:
|
||
severity_after = str(change.get("severity_after") or "none")
|
||
severity_before = str(change.get("severity_before") or "none")
|
||
severity_delta = max(severity_rank(severity_after) - severity_rank(severity_before), 0)
|
||
cases_delta = max(int(change.get("open_cases_delta") or 0), 0)
|
||
detections_delta = max(int(change.get("detections_delta") or 0), 0)
|
||
locks_delta = max(int(change.get("active_locks_delta") or 0), 0)
|
||
score_delta = max(int(change.get("score_delta") or 0), 0)
|
||
forecast_before = float(change.get("forecast_before") or 0)
|
||
forecast_delta = float(change.get("forecast_delta") or 0)
|
||
forecast_drop_pct = 0.0
|
||
if forecast_delta < 0:
|
||
forecast_drop_pct = abs(forecast_delta) / max(abs(forecast_before), 1.0) * 100.0
|
||
|
||
priority_score = round(
|
||
float(change.get("significance") or 0)
|
||
+ severity_delta * 24
|
||
+ cases_delta * 8
|
||
+ detections_delta * 5
|
||
+ locks_delta * 10
|
||
+ score_delta * 0.8
|
||
+ forecast_drop_pct * 0.6,
|
||
2,
|
||
)
|
||
|
||
reasons: list[str] = []
|
||
if severity_delta > 0:
|
||
reasons.append("рост severity")
|
||
if cases_delta > 0:
|
||
reasons.append(f"рост кейсов +{cases_delta}")
|
||
if detections_delta > 0:
|
||
reasons.append(f"рост detections +{detections_delta}")
|
||
if locks_delta > 0:
|
||
reasons.append(f"рост блокировок +{locks_delta}")
|
||
if forecast_drop_pct >= 10:
|
||
reasons.append(f"просадка прогноза {round(forecast_drop_pct, 1)}%")
|
||
if change.get("registry_match_mode") == "manual":
|
||
reasons.append("manual match")
|
||
|
||
if priority_score >= 140:
|
||
priority_tier = "critical"
|
||
elif priority_score >= 85:
|
||
priority_tier = "high"
|
||
elif priority_score >= 40:
|
||
priority_tier = "medium"
|
||
else:
|
||
priority_tier = "low"
|
||
|
||
change["priority_score"] = priority_score
|
||
change["priority_tier"] = priority_tier
|
||
change["priority_reason"] = ", ".join(reasons[:4]) if reasons else "слабый сдвиг без явного триггера"
|
||
return change
|
||
|
||
|
||
def load_previous_artifact(state_dir: Path) -> dict[str, Any] | None:
|
||
latest_path = state_dir / "latest.json"
|
||
if not latest_path.exists():
|
||
return None
|
||
try:
|
||
return json.loads(latest_path.read_text(encoding="utf-8"))
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
def snapshot_from_context(context: dict[str, Any]) -> dict[tuple[Any, Any], dict[str, Any]]:
|
||
snapshot_items = context.get("portfolio_snapshot") or []
|
||
if snapshot_items:
|
||
return {
|
||
(item.get("infobase"), item.get("company_entity_key") or item.get("counterparty")): item
|
||
for item in snapshot_items
|
||
}
|
||
|
||
merged: dict[tuple[Any, Any], dict[str, Any]] = {}
|
||
for source_name in ("top_risks", "top_forecasts", "watchlist", "busy_bases"):
|
||
for item in context.get(source_name, []):
|
||
key = (item.get("infobase"), item.get("company_entity_key") or item.get("counterparty"))
|
||
if key not in merged:
|
||
merged[key] = dict(item)
|
||
else:
|
||
merged[key].update({k: v for k, v in item.items() if v not in (None, "")})
|
||
return merged
|
||
|
||
|
||
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,
|
||
company_entity_key,
|
||
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,
|
||
company_entity_key,
|
||
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,
|
||
p.company_entity_key,
|
||
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.company_entity_key = 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,
|
||
company_entity_key,
|
||
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,
|
||
p.company_entity_key,
|
||
c.entity_id AS counterparty,
|
||
c.title,
|
||
c.severity,
|
||
c.status
|
||
FROM analytics_1c.cases AS c
|
||
LEFT JOIN analytics_1c.v_company_portfolio_overview AS p
|
||
ON p.infobase = c.infobase AND p.company_entity_key = c.entity_id
|
||
WHERE c.entity_type = 'counterparty' AND c.status != 'closed'
|
||
ORDER BY c.opened_at DESC
|
||
LIMIT {int(top_limit)}
|
||
"""
|
||
)
|
||
)
|
||
|
||
portfolio_snapshot = rows_to_dict(
|
||
client.query(
|
||
"""
|
||
SELECT
|
||
infobase,
|
||
company_entity_key,
|
||
counterparty,
|
||
normalized_counterparty,
|
||
registry_match_mode,
|
||
signal_severity,
|
||
signal_score,
|
||
current_status,
|
||
active_locks,
|
||
days_since_last_activity,
|
||
round(amount_30d, 2) AS amount_30d,
|
||
round(amount_forecast_30d, 2) AS amount_forecast_30d,
|
||
open_cases_total,
|
||
detections_total
|
||
FROM analytics_1c.v_company_portfolio_overview
|
||
ORDER BY counterparty, infobase
|
||
"""
|
||
)
|
||
)
|
||
|
||
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,
|
||
"portfolio_snapshot": portfolio_snapshot,
|
||
}
|
||
|
||
|
||
def compute_delta_context(current: dict[str, Any], previous_artifact: dict[str, Any] | None) -> dict[str, Any]:
|
||
if not previous_artifact:
|
||
return {
|
||
"available": False,
|
||
"reason": "no previous brief artifact",
|
||
"current_generated_at": current.get("generated_at"),
|
||
}
|
||
|
||
previous = previous_artifact.get("context", {})
|
||
current_summary = current.get("portfolio_summary", {})
|
||
previous_summary = previous.get("portfolio_summary", {})
|
||
current_watchlist = {(item.get("infobase"), item.get("company_entity_key") or item.get("counterparty")) for item in current.get("watchlist", [])}
|
||
previous_watchlist = {(item.get("infobase"), item.get("company_entity_key") or item.get("counterparty")) for item in previous.get("watchlist", [])}
|
||
current_snapshot = snapshot_from_context(current)
|
||
previous_snapshot = snapshot_from_context(previous)
|
||
|
||
delta_summary = {
|
||
"companies_total_delta": current_summary.get("companies_total", 0) - previous_summary.get("companies_total", 0),
|
||
"critical_total_delta": current_summary.get("critical_total", 0) - previous_summary.get("critical_total", 0),
|
||
"high_total_delta": current_summary.get("high_total", 0) - previous_summary.get("high_total", 0),
|
||
"busy_total_delta": current_summary.get("busy_total", 0) - previous_summary.get("busy_total", 0),
|
||
"open_cases_total_delta": current_summary.get("open_cases_total", 0) - previous_summary.get("open_cases_total", 0),
|
||
"detections_total_delta": current_summary.get("detections_total", 0) - previous_summary.get("detections_total", 0),
|
||
"activity_30d_total_delta": round(
|
||
float(current_summary.get("activity_30d_total", 0) or 0)
|
||
- float(previous_summary.get("activity_30d_total", 0) or 0),
|
||
2,
|
||
),
|
||
"activity_forecast_30d_total_delta": round(
|
||
float(current_summary.get("activity_forecast_30d_total", 0) or 0)
|
||
- float(previous_summary.get("activity_forecast_30d_total", 0) or 0),
|
||
2,
|
||
),
|
||
}
|
||
|
||
new_critical: list[str] = []
|
||
resolved_critical: list[str] = []
|
||
top_changes: list[dict[str, Any]] = []
|
||
|
||
for key, current_item in current_snapshot.items():
|
||
previous_item = previous_snapshot.get(key)
|
||
if not previous_item:
|
||
continue
|
||
|
||
current_severity = str(current_item.get("signal_severity") or "none")
|
||
previous_severity = str(previous_item.get("signal_severity") or "none")
|
||
current_rank = severity_rank(current_severity)
|
||
previous_rank = severity_rank(previous_severity)
|
||
score_before = int(previous_item.get("signal_score") or 0)
|
||
score_after = int(current_item.get("signal_score") or 0)
|
||
score_delta = score_after - score_before
|
||
cases_before = int(previous_item.get("open_cases_total") or 0)
|
||
cases_after = int(current_item.get("open_cases_total") or 0)
|
||
cases_delta = cases_after - cases_before
|
||
detections_before = int(previous_item.get("detections_total") or 0)
|
||
detections_after = int(current_item.get("detections_total") or 0)
|
||
detections_delta = detections_after - detections_before
|
||
locks_before = int(previous_item.get("active_locks") or 0)
|
||
locks_after = int(current_item.get("active_locks") or 0)
|
||
locks_delta = locks_after - locks_before
|
||
forecast_before = float(previous_item.get("amount_forecast_30d") or 0)
|
||
forecast_after = float(current_item.get("amount_forecast_30d") or 0)
|
||
forecast_delta = round(forecast_after - forecast_before, 2)
|
||
|
||
if current_severity == "critical" and previous_severity != "critical":
|
||
new_critical.append(str(current_item.get("counterparty") or "-"))
|
||
if previous_severity == "critical" and current_severity != "critical":
|
||
resolved_critical.append(str(current_item.get("counterparty") or "-"))
|
||
|
||
change_type = None
|
||
summary = None
|
||
significance = 0.0
|
||
|
||
if current_rank > previous_rank:
|
||
change_type = "severity_up"
|
||
summary = f"Severity {previous_severity} -> {current_severity}, score {score_before} -> {score_after}."
|
||
significance = max(significance, (current_rank - previous_rank) * 50 + max(score_delta, 0))
|
||
elif current_rank < previous_rank:
|
||
change_type = "severity_down"
|
||
summary = f"Severity {previous_severity} -> {current_severity}, напряжение по компании снизилось."
|
||
significance = max(significance, (previous_rank - current_rank) * 40 + abs(score_delta))
|
||
|
||
if cases_delta > 0 and cases_delta * 6 > significance:
|
||
change_type = "cases_up"
|
||
summary = f"Открытых кейсов стало больше: {cases_before} -> {cases_after}."
|
||
significance = cases_delta * 6 + max(score_delta, 0)
|
||
|
||
if locks_delta > 0 and locks_delta * 8 > significance:
|
||
change_type = "locks_up"
|
||
summary = f"Активные блокировки выросли: {locks_before} -> {locks_after}."
|
||
significance = locks_delta * 8 + max(score_delta, 0)
|
||
|
||
if forecast_delta < 0:
|
||
forecast_drop_pct = abs(forecast_delta) / max(abs(forecast_before), 1.0) * 100.0
|
||
if forecast_drop_pct > significance:
|
||
change_type = "forecast_drop"
|
||
summary = f"Прогноз активности 30д снизился: {round(forecast_before, 2)} -> {round(forecast_after, 2)}."
|
||
significance = forecast_drop_pct
|
||
elif forecast_delta > 0:
|
||
forecast_growth_pct = abs(forecast_delta) / max(abs(forecast_before), 1.0) * 100.0
|
||
if forecast_growth_pct > significance and not change_type:
|
||
change_type = "forecast_growth"
|
||
summary = f"Прогноз активности 30д вырос: {round(forecast_before, 2)} -> {round(forecast_after, 2)}."
|
||
significance = forecast_growth_pct
|
||
|
||
if detections_delta > 0 and detections_delta * 4 > significance:
|
||
change_type = "detections_up"
|
||
summary = f"Число detections выросло: {detections_before} -> {detections_after}."
|
||
significance = detections_delta * 4
|
||
|
||
if not change_type:
|
||
continue
|
||
|
||
top_changes.append(
|
||
rank_top_change(
|
||
{
|
||
"infobase": current_item.get("infobase"),
|
||
"company_entity_key": current_item.get("company_entity_key"),
|
||
"company": current_item.get("counterparty"),
|
||
"normalized_counterparty": current_item.get("normalized_counterparty"),
|
||
"registry_match_mode": current_item.get("registry_match_mode"),
|
||
"change_type": change_type,
|
||
"summary": summary,
|
||
"severity_before": previous_severity,
|
||
"severity_after": current_severity,
|
||
"score_before": score_before,
|
||
"score_after": score_after,
|
||
"score_delta": score_delta,
|
||
"open_cases_before": cases_before,
|
||
"open_cases_after": cases_after,
|
||
"open_cases_delta": cases_delta,
|
||
"detections_before": detections_before,
|
||
"detections_after": detections_after,
|
||
"detections_delta": detections_delta,
|
||
"active_locks_before": locks_before,
|
||
"active_locks_after": locks_after,
|
||
"active_locks_delta": locks_delta,
|
||
"forecast_before": round(forecast_before, 2),
|
||
"forecast_after": round(forecast_after, 2),
|
||
"forecast_delta": forecast_delta,
|
||
"significance": round(significance, 2),
|
||
}
|
||
)
|
||
)
|
||
|
||
entered_watchlist = sorted(
|
||
key[1] for key in current_watchlist - previous_watchlist if key[1]
|
||
)
|
||
left_watchlist = sorted(
|
||
key[1] for key in previous_watchlist - current_watchlist if key[1]
|
||
)
|
||
|
||
top_changes.sort(
|
||
key=lambda item: (
|
||
float(item.get("priority_score") or 0),
|
||
float(item.get("significance") or 0),
|
||
int(item.get("score_after") or 0),
|
||
int(item.get("open_cases_after") or 0),
|
||
),
|
||
reverse=True,
|
||
)
|
||
|
||
delta_summary.update(
|
||
{
|
||
"new_critical_total": len(new_critical),
|
||
"resolved_critical_total": len(resolved_critical),
|
||
"entered_watchlist_total": len(entered_watchlist),
|
||
"left_watchlist_total": len(left_watchlist),
|
||
}
|
||
)
|
||
|
||
return {
|
||
"available": True,
|
||
"previous_generated_at": previous.get("generated_at") or previous_artifact.get("generated_at"),
|
||
"current_generated_at": current.get("generated_at"),
|
||
"summary": delta_summary,
|
||
"new_critical": new_critical[:10],
|
||
"resolved_critical": resolved_critical[:10],
|
||
"entered_watchlist": entered_watchlist[:10],
|
||
"left_watchlist": left_watchlist[:10],
|
||
"top_changes": top_changes[:15],
|
||
}
|
||
|
||
|
||
def render_deterministic_payload(context: dict[str, Any]) -> dict[str, Any]:
|
||
summary = context["portfolio_summary"]
|
||
freshness = context["freshness"]
|
||
delta = context.get("delta", {})
|
||
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 delta.get("available"):
|
||
delta_summary = delta.get("summary", {})
|
||
summary_lines.append(
|
||
"С прошлого запуска: "
|
||
f"critical {delta_summary.get('critical_total_delta', 0):+d}, "
|
||
f"busy {delta_summary.get('busy_total_delta', 0):+d}, "
|
||
f"кейсы {delta_summary.get('open_cases_total_delta', 0):+d}, "
|
||
f"detections {delta_summary.get('detections_total_delta', 0):+d}."
|
||
)
|
||
if delta.get("top_changes"):
|
||
leaders = ", ".join(item["company"] for item in delta["top_changes"][:3] if item.get("company"))
|
||
if leaders:
|
||
summary_lines.append(f"Главные изменения с прошлого запуска: {leaders}.")
|
||
if stale_sources:
|
||
summary_lines.append(f"Есть просрочка по источникам: {', '.join(stale_sources)}.")
|
||
else:
|
||
summary_lines.append("Свежесть источников укладывается в заданный порог.")
|
||
|
||
primary_risk = top_risks[0] if top_risks else {}
|
||
primary_risk_company = primary_risk.get("counterparty") or "лидирующие компании риска"
|
||
primary_risk_reason = primary_risk.get("top_signal") or "рост operational severity"
|
||
activity_30d_total = float(summary.get("activity_30d_total") or 0)
|
||
activity_forecast_30d_total = float(summary.get("activity_forecast_30d_total") or 0)
|
||
if activity_30d_total > 0:
|
||
activity_ratio = activity_forecast_30d_total / activity_30d_total
|
||
else:
|
||
activity_ratio = 1.0
|
||
if activity_ratio < 0.95:
|
||
activity_outlook = "Есть риск спада активности в ближайшие 30 дней."
|
||
activity_effect = "Проверить, не перейдёт ли операционный спад в падение выручки и прибыли по топ-компаниям."
|
||
elif activity_ratio > 1.05:
|
||
activity_outlook = "Есть потенциал роста активности в ближайшие 30 дней."
|
||
activity_effect = "Важно не потерять рост из-за кейсов, блокировок и ручных несоответствий."
|
||
else:
|
||
activity_outlook = "По активности на 30 дней картина близка к стабильной."
|
||
activity_effect = "Резкого сигнала на рост или падение по активности сейчас нет."
|
||
primary_risk_load = []
|
||
if int(primary_risk.get("open_cases_total") or 0) > 0:
|
||
primary_risk_load.append(f"кейсы {int(primary_risk['open_cases_total'])}")
|
||
if int(primary_risk.get("detections_total") or 0) > 0:
|
||
primary_risk_load.append(f"detections {int(primary_risk['detections_total'])}")
|
||
if int(primary_risk.get("active_locks") or 0) > 0:
|
||
primary_risk_load.append(f"блокировки {int(primary_risk['active_locks'])}")
|
||
primary_risk_tail = f" ({', '.join(primary_risk_load)})" if primary_risk_load else ""
|
||
|
||
delta_summary = delta.get("summary", {}) if delta.get("available") else {}
|
||
if delta.get("available"):
|
||
change_answer = (
|
||
"С прошлого запуска "
|
||
f"critical {delta_summary.get('critical_total_delta', 0):+d}, "
|
||
f"busy {delta_summary.get('busy_total_delta', 0):+d}, "
|
||
f"кейсы {delta_summary.get('open_cases_total_delta', 0):+d}, "
|
||
f"detections {delta_summary.get('detections_total_delta', 0):+d}."
|
||
)
|
||
change_action = (
|
||
f"Сначала разобрать новые critical: {', '.join(delta.get('new_critical', [])[:3])}."
|
||
if delta.get("new_critical")
|
||
else "Проверить компании с самым большим ухудшением и подтвердить владельцев на закрытие."
|
||
)
|
||
else:
|
||
change_answer = "Сравнение с прошлым запуском сейчас недоступно, поэтому смотреть нужно на текущие кейсы, блокировки и свежесть источников."
|
||
change_action = "После следующего запуска включить сравнение и отслеживать не только уровень риска, но и его динамику."
|
||
|
||
trust_parts = []
|
||
if stale_sources:
|
||
trust_parts.append(f"есть просрочка по источникам: {', '.join(stale_sources)}")
|
||
if summary["manual_total"] > 0:
|
||
trust_parts.append(f"manual-match компаний: {summary['manual_total']}")
|
||
trust_answer = (
|
||
"Выводы рабочие, но требуют осторожности: " + "; ".join(trust_parts) + "."
|
||
if trust_parts
|
||
else "Выводам можно доверять как operational-картине: источники свежие, явных ограничений по данным нет."
|
||
)
|
||
trust_action = (
|
||
"Перед жёсткими оргвыводами вручную перепроверить manual-match и просроченные источники."
|
||
if trust_parts
|
||
else "Использовать brief как основу для управленческого разбора без дополнительного технического triage."
|
||
)
|
||
|
||
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 delta.get("available") and delta.get("summary", {}).get("new_critical_total", 0) > 0:
|
||
actions.insert(0, f"Сначала разобрать новые critical-компании: {', '.join(delta.get('new_critical', [])[:3])}.")
|
||
if watchlist:
|
||
actions[1] = (
|
||
f"Проверить watchlist: {', '.join(item['counterparty'] for item in watchlist[:3])}."
|
||
)
|
||
|
||
manager_questions = [
|
||
{
|
||
"question": "Что происходит сейчас?",
|
||
"answer": (
|
||
f"Портфель под операционным давлением: critical {summary['critical_total']}, "
|
||
f"high {summary['high_total']}, открытых кейсов {summary['open_cases_total']}."
|
||
),
|
||
"recommended_action": "Не распыляться: взять в первую очередь не больше 5 компаний с максимальной нагрузкой по кейсам и блокировкам.",
|
||
},
|
||
{
|
||
"question": "Где главный риск?",
|
||
"answer": (
|
||
f"Главный риск сейчас в {primary_risk_company}: {primary_risk_reason}{primary_risk_tail}."
|
||
if primary_risk
|
||
else "Главный риск не выделен: в данных нет явного лидера по severity."
|
||
),
|
||
"recommended_action": (
|
||
f"Сразу назначить владельца на {primary_risk_company} и проверить кейсы, блокировки и фактическую активность."
|
||
if primary_risk
|
||
else "Сначала вручную выделить 1-3 компании с самым высоким operational pressure."
|
||
),
|
||
},
|
||
{
|
||
"question": "Что изменилось с прошлого запуска?",
|
||
"answer": change_answer,
|
||
"recommended_action": change_action,
|
||
},
|
||
{
|
||
"question": "Есть ли риск падения прибыли в ближайшие 30 дней?",
|
||
"answer": (
|
||
"Прямого прогноза прибыли в этих данных нет. "
|
||
f"{activity_outlook} {activity_effect}"
|
||
),
|
||
"recommended_action": "Сверить top-10 компаний по активности с выручкой, маржой и прибылью, чтобы отделить operational-шум от реального финансового риска.",
|
||
},
|
||
{
|
||
"question": "Что делать сегодня?",
|
||
"answer": "Нужен короткий управленческий triage: подтвердить причину проблемы, владельца, срок и измеримый результат по каждой компании первой очереди.",
|
||
"recommended_action": actions[0] if actions else "Разобрать компании с максимальным signal score и открытыми кейсами.",
|
||
},
|
||
{
|
||
"question": "Насколько этим выводам можно доверять?",
|
||
"answer": trust_answer,
|
||
"recommended_action": trust_action,
|
||
},
|
||
]
|
||
|
||
management_plan = [
|
||
{
|
||
"horizon": "today",
|
||
"focus": "Остановить прирост проблем по компаниям первой очереди.",
|
||
"action": actions[0] if actions else "Разобрать компании с максимальным signal score и открытыми кейсами.",
|
||
"expected_effect": "Снять главный операционный тормоз и не допустить роста хвоста кейсов до следующего цикла.",
|
||
"metric": "open_cases_total и active_locks по 5 приоритетным компаниям к вечеру.",
|
||
},
|
||
{
|
||
"horizon": "week",
|
||
"focus": "Вернуть управляемость по спорным и зависшим предприятиям.",
|
||
"action": (
|
||
f"{actions[1]} {actions[2]}"
|
||
if len(actions) >= 3
|
||
else "Проверить watchlist, manual-match и причины повторного попадания компаний в риск."
|
||
),
|
||
"expected_effect": "Очистить ложный шум, снять ручные ошибки и сосредоточить команду на реальных потерях.",
|
||
"metric": "watchlist, manual-match и повторные critical по неделе.",
|
||
},
|
||
{
|
||
"horizon": "30d",
|
||
"focus": "Понять, где риск для прибыли, а где точка роста.",
|
||
"action": "Сверить top-10 компаний по активности и forecast с выручкой, маржой и прибылью; закрепить 3 предприятия, где снятие кейсов должно дать самый быстрый бизнес-эффект.",
|
||
"expected_effect": "Отделить операционный риск от финансового и направить усилия туда, где можно удержать или нарастить прибыль.",
|
||
"metric": "связка activity_forecast_30d_total с выручкой/маржой/прибылью по топ-компаниям.",
|
||
},
|
||
]
|
||
|
||
caveats = [
|
||
"Показатель amount здесь трактуется как activity score, а не как деньги или выручка.",
|
||
"Severity operational-driven: high/critical отражают кейсы, detections и занятость базы, а не автоматически финансовый риск.",
|
||
"Без прямых финансовых данных brief не должен трактоваться как готовый прогноз прибыли.",
|
||
]
|
||
if summary["manual_total"] > 0:
|
||
caveats.append("Компании с registry_match_mode=manual требуют осторожности при юридической интерпретации реестра.")
|
||
|
||
return {
|
||
"headline": headline,
|
||
"summary": summary_lines[:6],
|
||
"manager_questions": manager_questions[:6],
|
||
"management_plan": management_plan[:5],
|
||
"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.get("manager_questions", []), start=1):
|
||
lines.append(
|
||
f"{idx}. {item['question']} — {item['answer']} Действие: {item['recommended_action']}"
|
||
)
|
||
lines.extend(["", "## План действий руководителя"])
|
||
for idx, item in enumerate(payload.get("management_plan", []), start=1):
|
||
lines.append(
|
||
f"{idx}. [{item['horizon']}] {item['focus']} Действие: {item['action']} Эффект: {item['expected_effect']} Метрика: {item['metric']}"
|
||
)
|
||
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 normalize_brief_payload(payload: dict[str, Any], context: dict[str, Any]) -> dict[str, Any]:
|
||
fallback = render_deterministic_payload(context)
|
||
normalized = dict(payload)
|
||
|
||
manager_questions = [
|
||
dict(item) for item in normalized.get("manager_questions", []) if isinstance(item, dict)
|
||
]
|
||
if not manager_questions:
|
||
manager_questions = [dict(item) for item in fallback["manager_questions"]]
|
||
profit_question = next(
|
||
item for item in fallback["manager_questions"] if "прибыл" in item["question"].lower()
|
||
)
|
||
if not any(
|
||
any(marker in str(item.get("question", "")).lower() for marker in ("прибыл", "финанс"))
|
||
for item in manager_questions
|
||
):
|
||
if len(manager_questions) >= 6:
|
||
manager_questions[-1] = dict(profit_question)
|
||
else:
|
||
manager_questions.insert(min(3, len(manager_questions)), dict(profit_question))
|
||
normalized["manager_questions"] = manager_questions[:6]
|
||
|
||
management_plan = [
|
||
dict(item) for item in normalized.get("management_plan", []) if isinstance(item, dict)
|
||
]
|
||
by_horizon = {str(item.get("horizon")): item for item in management_plan if item.get("horizon")}
|
||
for item in fallback["management_plan"]:
|
||
by_horizon.setdefault(str(item["horizon"]), dict(item))
|
||
normalized["management_plan"] = [
|
||
by_horizon[horizon]
|
||
for horizon in ("today", "week", "30d")
|
||
if horizon in by_horizon
|
||
][:5]
|
||
|
||
caveats = [str(item) for item in normalized.get("caveats", []) if str(item).strip()]
|
||
profit_caveat = next(item for item in fallback["caveats"] if "прибыл" in item.lower())
|
||
if not any("прибыл" in item.lower() for item in caveats):
|
||
caveats.append(profit_caveat)
|
||
normalized["caveats"] = caveats[:4] or list(fallback["caveats"])
|
||
|
||
return normalized
|
||
|
||
|
||
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)
|
||
|
||
previous_artifact = load_previous_artifact(state_dir)
|
||
client = ch_client(args)
|
||
context = build_context(client, top_limit=args.top_limit, freshness_hours=args.freshness_hours)
|
||
context["delta"] = compute_delta_context(context, previous_artifact)
|
||
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"
|
||
|
||
payload = normalize_brief_payload(payload, context)
|
||
|
||
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())
|