#!/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 pathlib import Path from typing import Any from company_intelligence_api import build_weekly_trend_report, manager_brief_state_dir ROOT = Path(__file__).resolve().parents[1] PROMPT_PATH = ROOT / "ai" / "weekly_digest_prompt.md" SCHEMA_PATH = ROOT / "ai" / "weekly_digest_schema.json" def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="Generate weekly executive digest for analytics_1c") p.add_argument( "--brief-state-dir", default=os.getenv("AW_1C_MANAGER_BRIEF_STATE_DIR", str(manager_brief_state_dir())), ) p.add_argument( "--state-dir", default=os.getenv("AW_1C_WEEKLY_DIGEST_STATE_DIR", str(ROOT / "state" / "weekly-digest")), ) 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( "--timeout-sec", type=int, default=int(os.getenv("AW_1C_MANAGER_BRIEF_TIMEOUT_SEC", "300")), ) p.add_argument( "--days", type=int, default=int(os.getenv("AW_1C_WEEKLY_DIGEST_DAYS", "7")), ) return p.parse_args() 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 load_latest_brief(brief_state_dir: Path) -> dict[str, Any]: latest_path = brief_state_dir / "latest.json" if not latest_path.exists(): raise RuntimeError(f"missing latest manager brief: {latest_path}") return json.loads(latest_path.read_text(encoding="utf-8")) def load_brief_history_payloads(brief_state_dir: Path, limit: int = 400) -> list[dict[str, Any]]: history_dir = brief_state_dir / "history" if not history_dir.exists(): return [] 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")) payload["_path"] = path.name items.append(payload) return items def classify_improvements(payloads: list[dict[str, Any]], limit: int = 5) -> list[dict[str, Any]]: improvements: dict[tuple[str, str], dict[str, Any]] = {} for payload in payloads: delta = payload.get("context", {}).get("delta") or {} generated_at = payload.get("generated_at") for item in delta.get("top_changes", []): change_type = str(item.get("change_type") or "") score_delta = float(item.get("score_delta") or 0) open_cases_delta = int(item.get("open_cases_delta") or 0) forecast_delta = float(item.get("forecast_delta") or 0) improved = False signal = "" if change_type == "severity_down": improved = True signal = f"severity снизилась {item.get('severity_before')} -> {item.get('severity_after')}" elif change_type == "forecast_growth" and forecast_delta > 0: improved = True signal = f"прогноз 30д вырос на {round(forecast_delta, 2)}" elif score_delta < 0 or open_cases_delta < 0: improved = True signal = ( f"score Δ {int(score_delta)}" if score_delta < 0 else f"open cases Δ {open_cases_delta}" ) if not improved: continue key = (str(item.get("infobase") or ""), str(item.get("company") or "")) candidate = { "company": item.get("company"), "infobase": item.get("infobase"), "signal": signal, "meaning": str(item.get("summary") or "Нагрузка по компании ослабла."), "generated_at": generated_at, "severity_after": item.get("severity_after"), } existing = improvements.get(key) if existing is None: improvements[key] = candidate return list(improvements.values())[:limit] def build_context(brief_state_dir: Path, days: int) -> dict[str, Any]: latest_brief = load_latest_brief(brief_state_dir) history_payloads = load_brief_history_payloads(brief_state_dir, limit=400) weekly_report = build_weekly_trend_report(history_payloads, days=days) latest_context = latest_brief.get("context", {}) latest_summary = latest_context.get("portfolio_summary", {}) daily = weekly_report.get("daily", []) first_day = daily[0] if daily else {} last_day = daily[-1] if daily else {} week_delta = { "critical_total_delta": int(last_day.get("critical_total", 0) or 0) - int(first_day.get("critical_total", 0) or 0), "busy_total_delta": int(last_day.get("busy_total", 0) or 0) - int(first_day.get("busy_total", 0) or 0), "open_cases_total_delta": int(last_day.get("open_cases_total", 0) or 0) - int(first_day.get("open_cases_total", 0) or 0), "detections_total_delta": int(last_day.get("detections_total", 0) or 0) - int(first_day.get("detections_total", 0) or 0), "activity_30d_total_delta": round(float(last_day.get("activity_30d_total", 0) or 0) - float(first_day.get("activity_30d_total", 0) or 0), 2), "activity_forecast_30d_total_delta": round(float(last_day.get("activity_forecast_30d_total", 0) or 0) - float(first_day.get("activity_forecast_30d_total", 0) or 0), 2), } return { "generated_at": datetime.now(UTC).isoformat(), "period_start": weekly_report.get("period_start"), "period_end": weekly_report.get("period_end"), "days": days, "latest_summary": latest_summary, "freshness": latest_context.get("freshness", []), "top_weekly_changes": weekly_report.get("top_weekly_changes", []), "daily": daily, "week_delta": week_delta, "improvements": classify_improvements(history_payloads), "latest_brief_headline": latest_brief.get("brief", {}).get("headline", ""), } def render_deterministic_digest(context: dict[str, Any]) -> dict[str, Any]: latest = context.get("latest_summary", {}) week_delta = context.get("week_delta", {}) top_changes = context.get("top_weekly_changes", []) improvements = context.get("improvements", []) stale_sources = [item["source"] for item in context.get("freshness", []) if item.get("stale")] headline = ( f"Неделя: critical {latest.get('critical_total', 0)}, busy {latest.get('busy_total', 0)}, " f"кейсы {latest.get('open_cases_total', 0)}." ) summary = [ f"За неделю: critical {week_delta.get('critical_total_delta', 0):+d}, busy {week_delta.get('busy_total_delta', 0):+d}, кейсы {week_delta.get('open_cases_total_delta', 0):+d}, detections {week_delta.get('detections_total_delta', 0):+d}.", f"Текущий портфель: компаний {latest.get('companies_total', 0)}, direct {latest.get('direct_total', 0)}, alias {latest.get('alias_total', 0)}, manual {latest.get('manual_total', 0)}.", f"Активность 30д {latest.get('activity_30d_total', 0)}, прогноз 30д {latest.get('activity_forecast_30d_total', 0)}.", ] if stale_sources: summary.append(f"Есть просроченные источники: {', '.join(stale_sources)}.") else: summary.append("Источник данных по неделе свежий, контур не выглядит просроченным.") top_priorities = [] for item in top_changes[:5]: top_priorities.append( { "company": str(item.get("company") or "-"), "priority": str(item.get("priority_tier") or "low"), "reason": str(item.get("priority_reason") or item.get("summary") or "-"), "recommended_action": ( "Проверить открытые кейсы, detections и фактическую занятость базы." if int(item.get("open_cases_delta") or 0) > 0 or int(item.get("active_locks_delta") or 0) > 0 else "Проверить причину weekly-сдвига и подтвердить, что это не накопленный operational шум." ), } ) actions = [ "Сначала разбирать weekly priority critical/high, а не весь красный портфель подряд.", "Для компаний с ростом кейсов и блокировок подтвердить, это operational перегрузка или реальный бизнес-сбой.", "По manual-match компаниям не делать жёстких выводов без проверки реестровой привязки.", ] if top_priorities: actions.insert(0, f"Приоритет недели: {', '.join(item['company'] for item in top_priorities[:3])}.") caveats = [ "Активность здесь operational-driven и не равна деньгам или выручке.", "Weekly приоритет строится по изменениям сигналов, кейсов, блокировок и прогноза, а не по бухгалтерскому результату.", ] if latest.get("manual_total", 0): caveats.append("Manual-match компании требуют осторожности при юридической трактовке соответствия реестру.") return { "headline": headline, "summary": summary[:7], "top_priorities": top_priorities, "improvements": improvements[:5], "actions": actions[:6], "caveats": caveats[:5], } def render_markdown(payload: dict[str, Any], generated_at: str) -> str: lines = [ "# Weekly Executive Digest 1C", "", f"_Сформировано: {generated_at}_", "", "## Заголовок", payload["headline"], "", "## Кратко", ] for item in payload["summary"]: lines.append(f"- {item}") lines.extend(["", "## Приоритет недели"]) for idx, item in enumerate(payload["top_priorities"], start=1): lines.append(f"{idx}. {item['company']} [{item['priority']}] — {item['reason']} Действие: {item['recommended_action']}") lines.extend(["", "## Что улучшилось"]) for idx, item in enumerate(payload["improvements"], start=1): lines.append(f"{idx}. {item['company']} — {item['signal']}. {item['meaning']}") 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-weekly-digest-", 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 main() -> int: args = parse_args() state_dir = Path(args.state_dir) brief_state_dir = Path(args.brief_state_dir) context = build_context(brief_state_dir, args.days) prompt = build_prompt(context) render_mode = "deterministic" model = "deterministic" digest = render_deterministic_digest(context) codex_stdout = "" rc, stdout_stderr, reply = run_codex(prompt, args) codex_stdout = stdout_stderr if rc == 0 and reply: try: digest = json.loads(reply) render_mode = "codex" model = args.model except json.JSONDecodeError: pass generated_at = datetime.now(UTC).isoformat() markdown = render_markdown(digest, generated_at) payload = { "generated_at": generated_at, "render_mode": render_mode, "model": model, "context": context, "digest": digest, "markdown": markdown, "codex_stdout": codex_stdout, } latest_json = state_dir / "latest.json" latest_md = state_dir / "latest.md" history_json = state_dir / "history" / f"{generated_at.replace(':', '').replace('-', '').replace('+00:00', 'Z')}.json" history_md = state_dir / "history" / f"{generated_at.replace(':', '').replace('-', '').replace('+00:00', 'Z')}.md" write_json(latest_json, payload) write_text(latest_md, markdown) write_json(history_json, payload) write_text(history_md, markdown) print(json.dumps({"status": "ok", "generated_at": generated_at, "render_mode": render_mode}, ensure_ascii=False)) return 0 if __name__ == "__main__": raise SystemExit(main())