feat(1c): add weekly digest and hard priority pages
This commit is contained in:
igor04091968
@@ -0,0 +1,347 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user