#!/usr/bin/env python3 from __future__ import annotations import argparse import html import json import os from datetime import UTC, datetime from decimal import Decimal from pathlib import Path from typing import Any from urllib.parse import quote import clickhouse_connect import uvicorn from fastapi import FastAPI, HTTPException, Query from fastapi.responses import HTMLResponse, PlainTextResponse, Response def parse_args() -> argparse.Namespace: p = argparse.ArgumentParser(description="Read-only company intelligence API for analytics_1c") p.add_argument("--host", default=os.getenv("AW_1C_COMPANY_API_HOST", "127.0.0.1")) p.add_argument("--port", type=int, default=int(os.getenv("AW_1C_COMPANY_API_PORT", "8710"))) return p.parse_args() def q(value: str) -> str: return "'" + value.replace("'", "''") + "'" 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(): return clickhouse_connect.get_client( host=os.getenv("CLICKHOUSE_HOST", "localhost"), port=int(os.getenv("CLICKHOUSE_PORT", "8123")), username=os.getenv("CLICKHOUSE_USER", "default"), password=os.getenv("CLICKHOUSE_PASSWORD", ""), database=os.getenv("CLICKHOUSE_DB", "analytics_1c"), ) 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")) def load_brief_history_records(limit: int = 20) -> list[dict[str, Any]]: history_dir = manager_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")) 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, "brief": payload.get("brief", {}), } ) return items def load_brief_history_payloads(limit: int = 200) -> list[dict[str, Any]]: history_dir = manager_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 load_brief_history_record(name: str) -> dict[str, Any]: safe_name = Path(name).name if not safe_name.endswith(".json"): safe_name += ".json" path = manager_brief_state_dir() / "history" / safe_name if not path.exists(): raise HTTPException(status_code=404, detail="manager brief history record not found") return json.loads(path.read_text(encoding="utf-8")) def extract_delta(payload: dict[str, Any]) -> dict[str, Any]: delta = payload.get("context", {}).get("delta") if isinstance(delta, dict): return delta return {"available": False, "reason": "delta not present in artifact"} def priority_score_for_change(item: dict[str, Any]) -> float: if item.get("priority_score") is not None: try: return float(item.get("priority_score")) except (TypeError, ValueError): return 0.0 score = 0.0 score += max(float(item.get("significance") or 0), 0) score += max(float(item.get("score_delta") or 0), 0) * 0.8 score += max(float(item.get("open_cases_delta") or 0), 0) * 8 score += max(float(item.get("detections_delta") or 0), 0) * 5 score += max(float(item.get("active_locks_delta") or 0), 0) * 10 return round(score, 2) def build_weekly_trend_report(payloads: list[dict[str, Any]], days: int = 7) -> dict[str, Any]: now = datetime.now(UTC) period_start = now.date().toordinal() - max(days - 1, 0) filtered: list[dict[str, Any]] = [] for payload in payloads: generated_at_raw = payload.get("generated_at") if not generated_at_raw: continue try: generated_at = datetime.fromisoformat(str(generated_at_raw)) except ValueError: continue if generated_at.tzinfo is None: generated_at = generated_at.replace(tzinfo=UTC) if generated_at.date().toordinal() < period_start: continue payload["_generated_dt"] = generated_at filtered.append(payload) by_day: dict[str, dict[str, Any]] = {} for payload in filtered: key = payload["_generated_dt"].date().isoformat() previous = by_day.get(key) if previous is None or payload["_generated_dt"] > previous["_generated_dt"]: by_day[key] = payload daily_rows: list[dict[str, Any]] = [] previous_summary: dict[str, Any] | None = None for key in sorted(by_day.keys()): payload = by_day[key] summary = payload.get("context", {}).get("portfolio_summary", {}) row = { "date": key, "generated_at": payload.get("generated_at"), "companies_total": int(summary.get("companies_total", 0) or 0), "critical_total": int(summary.get("critical_total", 0) or 0), "high_total": int(summary.get("high_total", 0) or 0), "busy_total": int(summary.get("busy_total", 0) or 0), "open_cases_total": int(summary.get("open_cases_total", 0) or 0), "detections_total": int(summary.get("detections_total", 0) or 0), "activity_30d_total": float(summary.get("activity_30d_total", 0) or 0), "activity_forecast_30d_total": float(summary.get("activity_forecast_30d_total", 0) or 0), } if previous_summary: row["critical_delta_vs_prev_day"] = row["critical_total"] - int(previous_summary.get("critical_total", 0) or 0) row["busy_delta_vs_prev_day"] = row["busy_total"] - int(previous_summary.get("busy_total", 0) or 0) row["open_cases_delta_vs_prev_day"] = row["open_cases_total"] - int(previous_summary.get("open_cases_total", 0) or 0) row["detections_delta_vs_prev_day"] = row["detections_total"] - int(previous_summary.get("detections_total", 0) or 0) else: row["critical_delta_vs_prev_day"] = 0 row["busy_delta_vs_prev_day"] = 0 row["open_cases_delta_vs_prev_day"] = 0 row["detections_delta_vs_prev_day"] = 0 daily_rows.append(row) previous_summary = summary weekly_changes: dict[tuple[str, str], dict[str, Any]] = {} for payload in filtered: delta = extract_delta(payload) for item in delta.get("top_changes", []): key = (str(item.get("infobase") or ""), str(item.get("company") or "")) candidate = dict(item) candidate["generated_at"] = payload.get("generated_at") candidate["priority_score"] = priority_score_for_change(candidate) existing = weekly_changes.get(key) if existing is None or float(candidate["priority_score"]) > float(existing.get("priority_score") or 0): weekly_changes[key] = candidate weekly_top_changes = sorted( weekly_changes.values(), key=lambda item: ( float(item.get("priority_score") or 0), float(item.get("significance") or 0), int(item.get("open_cases_delta") or 0), ), reverse=True, )[:20] latest = daily_rows[-1] if daily_rows else None return { "days": days, "period_start": datetime.fromordinal(period_start).date().isoformat(), "period_end": now.date().isoformat(), "daily": daily_rows, "latest": latest, "top_weekly_changes": weekly_top_changes, } def grafana_company_dashboard_url() -> str: return os.getenv( "AW_1C_MANAGER_BRIEF_GRAFANA_URL", "http://10.10.10.11:3000/d/1c-file-companies/1c-file-company-intelligence", ) def fmt_number(value: Any) -> str: if value is None or value == "": return "-" if isinstance(value, float): return f"{value:,.2f}".replace(",", " ").replace(".", ",") if isinstance(value, int): return f"{value:,}".replace(",", " ") return str(value) def severity_badge(severity: str) -> str: tone = { "critical": "critical", "high": "high", "medium": "medium", "low": "low", "none": "none", }.get((severity or "").lower(), "none") return f'{html.escape(severity or "none")}' def company_detail_url(counterparty: str, infobase: str | None = None) -> str: base = f"/manager/company/{quote(counterparty)}" if infobase: return f"{base}?infobase={quote(infobase)}" return base def manager_brief_history_html_url(name: str) -> str: safe_name = Path(name).name return f"/manager/briefs/{quote(safe_name)}" def render_manager_brief_html(payload: dict[str, Any]) -> str: brief = payload.get("brief", {}) context = payload.get("context", {}) summary = context.get("portfolio_summary", {}) freshness = context.get("freshness", []) top_risks = brief.get("top_risks", []) top_forecasts = brief.get("top_forecasts", []) actions = brief.get("actions", []) caveats = brief.get("caveats", []) render_mode = payload.get("render_mode", "unknown") generated_at = payload.get("generated_at", "") history_url = "/api/1/analytics-1c/manager/brief/history" history_html_url = "/manager/briefs" delta_html_url = "/manager/changes" weekly_html_url = "/manager/trends/weekly" problematic_1d_url = "/manager/problematic?days=1" problematic_7d_url = "/manager/problematic?days=7" json_url = "/api/1/analytics-1c/manager/brief/latest" md_url = "/api/1/analytics-1c/manager/brief/latest.md" grafana_url = grafana_company_dashboard_url() freshness_rows = [] for item in freshness: status = "stale" if item.get("stale") else "fresh" freshness_rows.append( "" f"{html.escape(str(item.get('source', '')))}" f"{html.escape(str(item.get('latest_ts', '-')))}" f"{html.escape(str(item.get('lag_hours', '-')))}" f"{'просрочен' if status == 'stale' else 'свежий'}" "" ) risk_cards = [] for item in top_risks: risk_cards.append( "
" f"

{html.escape(item.get('company', '-'))}

{severity_badge(item.get('severity', ''))}
" f"

{html.escape(item.get('reason', '-'))}

" f"

Действие: {html.escape(item.get('recommended_action', '-'))}

" f"

Открыть карточку компании

" "
" ) forecast_cards = [] for item in top_forecasts: company = item.get("company", "-") forecast_cards.append( "
" f"

{html.escape(company)}

" f"

Прогноз 30д: {html.escape(item.get('forecast_30d', '-'))}

" f"

{html.escape(item.get('interpretation', '-'))}

" f"Карточка компании" "
" ) summary_items = "\n".join(f"
  • {html.escape(str(item))}
    • " for item in brief.get("summary", [])) action_items = "\n".join(f"
  • {html.escape(str(item))}
    • " for item in actions) caveat_items = "\n".join(f"
  • {html.escape(str(item))}
    • " for item in caveats) return f""" 1C Executive Brief
    AW-rus · 1C Executive Brief · render mode: {html.escape(render_mode)}

    {html.escape(brief.get("headline", "Executive brief недоступен"))}

    Сформировано: {html.escape(generated_at)}

    Краткий комментарий

      {summary_items}

    Портфель

    Компаний
    {fmt_number(summary.get("companies_total"))}
    Critical
    {fmt_number(summary.get("critical_total"))}
    Кейсы
    {fmt_number(summary.get("open_cases_total"))}
    Прогноз 30д
    {fmt_number(summary.get("activity_forecast_30d_total"))}

    Компании риска

    {''.join(risk_cards) or '

    Нет данных.

    '}

    Прогноз активности 30 дней

    {''.join(forecast_cards) or '

    Нет данных.

    '}

    Что проверить руководителю

      {action_items}

    Ограничения интерпретации

      {caveat_items}

    Свежесть источников

    {''.join(freshness_rows)}
    Источник Последняя отметка Отставание, ч Статус
    """ def problematic_companies(days: int = 7, limit: int = 50) -> list[dict[str, Any]]: client = ch_client() sql = f""" WITH recent AS ( SELECT infobase, counterparty, max(generated_at) AS latest_signal_at, max(score) AS max_score, sum(score) AS total_score, count() AS signals_total, countIf(severity = 'critical') AS critical_total, countIf(severity = 'high') AS high_total, argMax(severity, tuple(score, generated_at)) AS top_severity, argMax(signal_type, tuple(score, generated_at)) AS top_signal_type, argMax(summary, tuple(score, generated_at)) AS top_summary FROM analytics_1c.company_health_signals WHERE generated_at >= now() - INTERVAL {int(days)} DAY GROUP BY infobase, counterparty ) SELECT p.infobase AS infobase, p.counterparty AS counterparty, p.company_name, p.normalized_counterparty, p.registry_match_mode, p.registry_assignee_name, p.registry_status, p.signal_severity, p.signal_score, p.amount_30d, p.amount_forecast_30d, p.current_status, p.active_locks, p.open_cases_total, p.detections_total, r.latest_signal_at, r.max_score, r.total_score, r.signals_total, r.critical_total, r.high_total, r.top_severity, r.top_signal_type, r.top_summary FROM recent AS r INNER JOIN analytics_1c.v_company_portfolio_overview AS p ON p.infobase = r.infobase AND p.counterparty = r.counterparty ORDER BY r.max_score DESC, r.signals_total DESC, p.amount_30d DESC, p.counterparty LIMIT {int(limit)} """ items = rows_to_dict(client.query(sql)) for item in items: if "infobase" not in item and "p.infobase" in item: item["infobase"] = item["p.infobase"] if "counterparty" not in item and "p.counterparty" in item: item["counterparty"] = item["p.counterparty"] return items def render_brief_history_html(items: list[dict[str, Any]]) -> str: rows = [] for item in items: rows.append( "" f"{html.escape(str(item.get('generated_at', '-')))}" f"{html.escape(str(item.get('render_mode', '-')))}" f"{html.escape(str(item.get('model') or '-'))}" f"{html.escape(str(item.get('headline') or '-'))}" f"Изменения" f"JSON" "" ) return f""" 1C Brief History

    История executive brief

    Текущий brief Что изменилось Неделя Проблемные 1д Проблемные 7д
    {''.join(rows) or ''}
    Сформировано Режим Модель Headline Delta Raw
    История пока пуста.
    """ def render_problematic_companies_html(items: list[dict[str, Any]], days: int) -> str: rows = [] for item in items: rows.append( "" f"{html.escape(str(item.get('counterparty', '-')))}" f"{html.escape(str(item.get('normalized_counterparty') or '-'))}" f"{severity_badge(str(item.get('top_severity') or item.get('signal_severity') or 'none'))}" f"{fmt_number(item.get('max_score'))}" f"{fmt_number(item.get('signals_total'))}" f"{fmt_number(item.get('critical_total'))}" f"{fmt_number(item.get('amount_30d'))}" f"{fmt_number(item.get('amount_forecast_30d'))}" f"{html.escape(str(item.get('top_signal_type') or '-'))}" f"{html.escape(str(item.get('top_summary') or '-'))}" "" ) return f""" 1C Problem Companies

    Проблемные компании за {days} {('день' if days == 1 else 'дней')}

    Список собран по live company signals. В приоритете max score, плотность сигналов и общий вес проблемного контура.

    Текущий brief История brief Неделя Срез 1д Срез 7д
    {''.join(rows) or ''}
    Компания Нормализация Severity Max score Signals Critical Активность 30д Прогноз 30д Top signal Комментарий
    Нет сигналов за выбранный период.
    """ def render_brief_delta_html(payload: dict[str, Any]) -> str: delta = extract_delta(payload) brief = payload.get("brief", {}) if not delta.get("available"): return f""" 1C Brief Changes

    Что изменилось с прошлого запуска

    Delta пока недоступна: {html.escape(str(delta.get('reason', 'unknown')))}.

    Вернуться к brief

    """ summary = delta.get("summary", {}) top_changes = delta.get("top_changes", []) new_critical = delta.get("new_critical", []) resolved_critical = delta.get("resolved_critical", []) entered_watchlist = delta.get("entered_watchlist", []) left_watchlist = delta.get("left_watchlist", []) def delta_value(value: Any) -> str: try: numeric = float(value) except (TypeError, ValueError): return html.escape(str(value)) prefix = "+" if numeric > 0 else "" if numeric.is_integer(): return prefix + str(int(numeric)) return prefix + f"{numeric:.2f}".replace(".", ",") stat_cards = [ ("Critical", delta_value(summary.get("critical_total_delta", 0))), ("Busy", delta_value(summary.get("busy_total_delta", 0))), ("Кейсы", delta_value(summary.get("open_cases_total_delta", 0))), ("Detections", delta_value(summary.get("detections_total_delta", 0))), ("Активность 30д", delta_value(summary.get("activity_30d_total_delta", 0))), ("Прогноз 30д", delta_value(summary.get("activity_forecast_30d_total_delta", 0))), ] stat_html = "".join( f"
    {html.escape(label)}
    {html.escape(value)}
    " for label, value in stat_cards ) tier_labels = { "critical": "Критический", "high": "Высокий", "medium": "Средний", "low": "Низкий", } rows = [] for item in top_changes: infobase = item.get("infobase") counterparty = item.get("company") priority_tier = str(item.get("priority_tier") or "low") rows.append( "" f"{html.escape(str(counterparty or '-'))}" f"{html.escape(tier_labels.get(priority_tier, priority_tier))}" f"{delta_value(item.get('priority_score', 0))}" f"{html.escape(str(item.get('change_type') or '-'))}" f"{html.escape(str(item.get('severity_before') or '-'))} -> {html.escape(str(item.get('severity_after') or '-'))}" f"{delta_value(item.get('score_delta', 0))}" f"{delta_value(item.get('open_cases_delta', 0))}" f"{delta_value(item.get('active_locks_delta', 0))}" f"{delta_value(item.get('forecast_delta', 0))}" f"{html.escape(str(item.get('priority_reason') or '-'))}" f"{html.escape(str(item.get('summary') or '-'))}" "" ) def as_list(items: list[str]) -> str: if not items: return "
  • Нет
    • " return "".join(f"
  • {html.escape(str(item))}
    • " for item in items) return f""" 1C Brief Changes

    Что изменилось с прошлого запуска

    Сравнение запусков: {html.escape(str(delta.get('previous_generated_at') or '-'))} -> {html.escape(str(delta.get('current_generated_at') or payload.get('generated_at') or '-'))}.

    Текущий brief История brief Неделя Проблемные 1д Проблемные 7д

    Сводка изменений

    {stat_html}

    Новые critical

      {as_list(new_critical)}

    Вышли из critical

      {as_list(resolved_critical)}

    Зашли в watchlist

      {as_list(entered_watchlist)}

    Вышли из watchlist

      {as_list(left_watchlist)}

    Ключевые изменения

    {''.join(rows) or ''}
    Компания Приоритет Score Тип Severity Score Δ Cases Δ Locks Δ Forecast Δ Причина приоритета Комментарий
    Нет выраженных изменений.
    """ def render_weekly_trend_html(report: dict[str, Any]) -> str: daily = report.get("daily", []) latest = report.get("latest") or {} top_weekly_changes = report.get("top_weekly_changes", []) def value(v: Any) -> str: return fmt_number(v) trend_rows = [] for item in daily: trend_rows.append( "" f"{html.escape(str(item.get('date', '-')))}" f"{value(item.get('companies_total'))}" f"{value(item.get('critical_total'))}" f"{value(item.get('busy_total'))}" f"{value(item.get('open_cases_total'))}" f"{value(item.get('detections_total'))}" f"{value(item.get('activity_30d_total'))}" f"{value(item.get('activity_forecast_30d_total'))}" f"{value(item.get('critical_delta_vs_prev_day'))}" f"{value(item.get('open_cases_delta_vs_prev_day'))}" "" ) change_rows = [] for item in top_weekly_changes: infobase = item.get("infobase") company = item.get("company") change_rows.append( "" f"{html.escape(str(company or '-'))}" f"{html.escape(str(item.get('priority_tier') or '-'))}" f"{value(item.get('priority_score'))}" f"{html.escape(str(item.get('change_type') or '-'))}" f"{value(item.get('open_cases_delta'))}" f"{value(item.get('active_locks_delta'))}" f"{value(item.get('forecast_delta'))}" f"{html.escape(str(item.get('priority_reason') or '-'))}" "" ) stat_cards = [ ("Компаний", value(latest.get("companies_total"))), ("Critical", value(latest.get("critical_total"))), ("Busy", value(latest.get("busy_total"))), ("Кейсы", value(latest.get("open_cases_total"))), ("Detections", value(latest.get("detections_total"))), ("Прогноз 30д", value(latest.get("activity_forecast_30d_total"))), ] stat_html = "".join( f"
    {html.escape(label)}
    {html.escape(val)}
    " for label, val in stat_cards ) return f""" 1C Weekly Trends

    Недельный тренд портфеля

    Период: {html.escape(str(report.get('period_start', '-')))} -> {html.escape(str(report.get('period_end', '-')))}. Страница показывает тренд по истории executive brief и недельный рейтинг проблемных компаний.

    Текущий brief Что изменилось История brief Проблемные 7д

    Текущее состояние по последнему дневному срезу

    {stat_html}

    Дневной тренд

    {''.join(trend_rows) or ''}
    Дата Компаний Critical Busy Кейсы Detections Активность 30д Прогноз 30д Critical Δ Cases Δ
    Недостаточно истории для недельного тренда.

    Недельный рейтинг приоритетов

    {''.join(change_rows) or ''}
    Компания Приоритет Priority score Тип Cases Δ Locks Δ Forecast Δ Причина
    За выбранный период заметных изменений нет.
    """ app = FastAPI(title="AW-rus 1C Company Intelligence API", version="1.0.0") @app.get("/favicon.ico") def favicon() -> Response: return Response(status_code=204) @app.get("/health") def health() -> dict[str, Any]: client = ch_client() summary = rows_to_dict( client.query( """ SELECT countIf(counterparty != '') AS documents_with_counterparty, (SELECT count() FROM analytics_1c.companies) AS companies_total, (SELECT count() FROM analytics_1c.company_registry) AS registry_rows_total, (SELECT count() FROM analytics_1c.company_forecasts) AS forecasts_total, (SELECT count() FROM analytics_1c.company_health_signals) AS health_signals_total FROM analytics_1c.documents """ ) )[0] return {"status": "ok", "generated_at": datetime.now(UTC).isoformat(), **summary} @app.get("/api/1/analytics-1c/companies/overview") def companies_overview( infobase: str | None = None, min_signal_score: int = Query(default=0, ge=0, le=100), limit: int = Query(default=50, ge=1, le=500), ) -> dict[str, Any]: client = ch_client() where = [f"signal_score >= {int(min_signal_score)}"] if infobase: where.append(f"infobase = {q(infobase)}") sql = f""" SELECT infobase, organization, counterparty, company_name, normalized_counterparty, registry_match_mode, registry_assignee_name, registry_status, registry_share_text, registry_key_contour, registry_inn, registry_kpp, owner_user, base_path, current_status, db_size_bytes, reglog_size_bytes, active_locks, current_activity_score, last_seen_at, days_since_last_activity, docs_7d, amount_7d, docs_30d, amount_30d, amount_forecast_30d, docs_forecast_30d, signal_severity, signal_score, top_signal FROM analytics_1c.v_company_portfolio_overview WHERE {' AND '.join(where)} ORDER BY signal_score DESC, amount_30d DESC, counterparty LIMIT {int(limit)} """ rows = rows_to_dict(client.query(sql)) return {"items": rows, "count": len(rows)} @app.get("/api/1/analytics-1c/companies/{counterparty}/summary") def company_summary(counterparty: str, infobase: str | None = None) -> dict[str, Any]: client = ch_client() filters = [f"counterparty = {q(counterparty)}"] if infobase: filters.append(f"infobase = {q(infobase)}") sql = f""" SELECT * FROM analytics_1c.v_company_portfolio_overview WHERE {' AND '.join(filters)} ORDER BY last_company_snapshot_at DESC, amount_30d DESC LIMIT 1 """ rows = rows_to_dict(client.query(sql)) if not rows: raise HTTPException(status_code=404, detail="counterparty not found in analytics_1c.v_company_portfolio_overview") card = rows[0] forecast_sql = f""" SELECT metric, horizon_days, baseline_daily, trend_slope, predicted_daily, predicted_total, confidence, note FROM analytics_1c.v_company_forecasts_current WHERE counterparty = {q(counterparty)} {"AND infobase = " + q(infobase) if infobase else ""} ORDER BY metric, horizon_days """ signals_sql = f""" SELECT generated_at, severity, score, signal_type, summary FROM analytics_1c.v_company_health_current WHERE counterparty = {q(counterparty)} {"AND infobase = " + q(infobase) if infobase else ""} ORDER BY score DESC, generated_at DESC """ timeline_sql = f""" SELECT last_company_snapshot_at AS ts, infobase, company_name, owner_user, current_status, db_size_bytes, reglog_size_bytes, active_locks, current_activity_score FROM analytics_1c.v_company_portfolio_overview WHERE counterparty = {q(counterparty)} {"AND infobase = " + q(infobase) if infobase else ""} ORDER BY ts DESC LIMIT 1 """ forecasts = rows_to_dict(client.query(forecast_sql)) signals = rows_to_dict(client.query(signals_sql)) company_state = rows_to_dict(client.query(timeline_sql)) timeline_sql = f""" SELECT ts, infobase, doc_type, operation_type, amount, status, author FROM analytics_1c.documents WHERE counterparty = {q(counterparty)} {"AND infobase = " + q(infobase) if infobase else ""} ORDER BY ts DESC LIMIT 20 """ timeline = rows_to_dict(client.query(timeline_sql)) essence = ( f"Компания {counterparty}: за 30 дней событий {card['docs_30d']}, суммарная активность {card['amount_30d']}, " f"прогноз активности на 30 дней {card['amount_forecast_30d']}, риск {card['signal_severity']}." ) return { "essence": essence, "card": card, "company_state": company_state[0] if company_state else None, "forecasts": forecasts, "signals": signals, "recent_documents": timeline, } @app.get("/api/1/analytics-1c/companies/{counterparty}/forecast") def company_forecast( counterparty: str, infobase: str | None = None, horizon_days: int | None = Query(default=None, ge=1, le=365), ) -> dict[str, Any]: client = ch_client() filters = [f"counterparty = {q(counterparty)}"] if infobase: filters.append(f"infobase = {q(infobase)}") if horizon_days is not None: filters.append(f"horizon_days = {int(horizon_days)}") sql = f""" SELECT generated_at, infobase, counterparty, metric, horizon_days, baseline_daily, trend_slope, predicted_daily, predicted_total, confidence, note FROM analytics_1c.v_company_forecasts_current WHERE {' AND '.join(filters)} ORDER BY metric, horizon_days """ rows = rows_to_dict(client.query(sql)) return {"items": rows, "count": len(rows)} @app.get("/api/1/analytics-1c/companies/{counterparty}/timeline") def company_timeline( counterparty: str, infobase: str | None = None, limit: int = Query(default=100, ge=1, le=500), ) -> dict[str, Any]: client = ch_client() filters = [f"counterparty = {q(counterparty)}"] if infobase: filters.append(f"infobase = {q(infobase)}") sql = f""" SELECT ts, infobase, organization, doc_type, doc_number, author, operation_type, amount, status, posted FROM analytics_1c.documents WHERE {' AND '.join(filters)} ORDER BY ts DESC LIMIT {int(limit)} """ rows = rows_to_dict(client.query(sql)) 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]: items = load_brief_history_records(limit) return {"items": items, "count": len(items)} @app.get("/api/1/analytics-1c/manager/brief/history/{name}") def manager_brief_history_record(name: str) -> dict[str, Any]: return load_brief_history_record(name) @app.get("/api/1/analytics-1c/manager/brief/delta/latest") def manager_brief_delta_latest() -> dict[str, Any]: payload = load_latest_manager_brief() delta = extract_delta(payload) return { "generated_at": payload.get("generated_at"), "headline": payload.get("brief", {}).get("headline"), "delta": delta, } @app.get("/api/1/analytics-1c/manager/trends/weekly") def manager_weekly_trends(days: int = Query(default=7, ge=2, le=30)) -> dict[str, Any]: return build_weekly_trend_report(load_brief_history_payloads(limit=400), days=days) @app.get("/api/1/analytics-1c/companies/problematic") def problematic_companies_api( days: int = Query(default=7, ge=1, le=30), limit: int = Query(default=50, ge=1, le=500), ) -> dict[str, Any]: items = problematic_companies(days=days, limit=limit) return {"items": items, "count": len(items), "days": days} @app.get("/manager/brief", response_class=HTMLResponse) def manager_brief_view() -> str: return render_manager_brief_html(load_latest_manager_brief()) @app.get("/manager/changes", response_class=HTMLResponse) def manager_brief_delta_view() -> str: return render_brief_delta_html(load_latest_manager_brief()) @app.get("/manager/trends/weekly", response_class=HTMLResponse) def manager_weekly_trends_view(days: int = Query(default=7, ge=2, le=30)) -> str: return render_weekly_trend_html(build_weekly_trend_report(load_brief_history_payloads(limit=400), days=days)) @app.get("/manager/briefs", response_class=HTMLResponse) def manager_brief_history_view(limit: int = Query(default=40, ge=1, le=200)) -> str: return render_brief_history_html(load_brief_history_records(limit)) @app.get("/manager/briefs/{name}", response_class=HTMLResponse) def manager_brief_history_detail_view(name: str) -> str: return render_manager_brief_html(load_brief_history_record(name)) @app.get("/manager/briefs/{name}/changes", response_class=HTMLResponse) def manager_brief_history_delta_view(name: str) -> str: return render_brief_delta_html(load_brief_history_record(name)) @app.get("/manager/problematic", response_class=HTMLResponse) def manager_problematic_companies_view( days: int = Query(default=7, ge=1, le=30), limit: int = Query(default=50, ge=1, le=500), ) -> str: return render_problematic_companies_html(problematic_companies(days=days, limit=limit), days) def render_company_detail_html(summary_payload: dict[str, Any], infobase: str | None = None) -> str: card = summary_payload["card"] company_state = summary_payload.get("company_state") or {} forecasts = summary_payload.get("forecasts") or [] signals = summary_payload.get("signals") or [] recent_documents = summary_payload.get("recent_documents") or [] title = card.get("counterparty", "Карточка компании") subtitle = summary_payload.get("essence", "") grafana_url = grafana_company_dashboard_url() summary_url = f"/api/1/analytics-1c/companies/{quote(card['counterparty'])}/summary" if infobase: summary_url += f"?infobase={quote(infobase)}" timeline_url = f"/api/1/analytics-1c/companies/{quote(card['counterparty'])}/timeline" if infobase: timeline_url += f"?infobase={quote(infobase)}" forecast_url = f"/api/1/analytics-1c/companies/{quote(card['counterparty'])}/forecast" if infobase: forecast_url += f"?infobase={quote(infobase)}" if card.get("signal_severity") == "critical": manager_comment = "Компания требует немедленного внимания: контур считает её operational-critical." elif card.get("signal_severity") == "high": manager_comment = "Компания в зоне повышенного внимания: нужен короткий управленческий разбор причин сигнала." elif card.get("signal_severity") == "medium": manager_comment = "Компания не аварийная, но требует точечной проверки причин отклонения." else: manager_comment = "По компании нет выраженного аварийного сигнала; смотреть контекст и тренд активности." metrics = [ ("Активность 7д", fmt_number(card.get("amount_7d"))), ("Активность 30д", fmt_number(card.get("amount_30d"))), ("Прогноз 30д", fmt_number(card.get("amount_forecast_30d"))), ("Документы 30д", fmt_number(card.get("docs_30d"))), ("Кейсы", fmt_number(card.get("open_cases_total"))), ("Detections", fmt_number(card.get("detections_total"))), ] metric_cards = "".join( f"
    {html.escape(label)}
    {html.escape(value)}
    " for label, value in metrics ) forecast_rows = [] for item in forecasts: forecast_rows.append( "" f"{html.escape(str(item.get('metric', '-')))}" f"{fmt_number(item.get('horizon_days'))}" f"{fmt_number(item.get('baseline_daily'))}" f"{fmt_number(item.get('predicted_total'))}" f"{fmt_number(item.get('confidence'))}" f"{html.escape(str(item.get('note', '-')))}" "" ) signal_rows = [] for item in signals[:12]: signal_rows.append( "" f"{html.escape(str(item.get('generated_at', '-')))}" f"{severity_badge(str(item.get('severity', 'none')))}" f"{fmt_number(item.get('score'))}" f"{html.escape(str(item.get('signal_type', '-')))}" f"{html.escape(str(item.get('summary', '-')))}" "" ) document_rows = [] for item in recent_documents[:20]: document_rows.append( "" f"{html.escape(str(item.get('ts', '-')))}" f"{html.escape(str(item.get('doc_type', '-')))}" f"{html.escape(str(item.get('operation_type', '-')))}" f"{fmt_number(item.get('amount'))}" f"{html.escape(str(item.get('status', '-')))}" f"{html.escape(str(item.get('author', '-')))}" "" ) registry_comment = ( f"Сопоставление с реестром: {card.get('registry_match_mode', 'none')}. " f"Ответственный: {card.get('registry_assignee_name') or 'не указан'}." ) if card.get("registry_match_mode") == "manual": registry_comment += " Требуется осторожность: запись заведена через manual override." return f""" {html.escape(title)} · 1C Company Brief
    AW-rus · Company Brief · {severity_badge(str(card.get("signal_severity", "none")))}

    {html.escape(title)}

    {html.escape(subtitle)}

    Управленческий комментарий

    {html.escape(manager_comment)}

    {html.escape(registry_comment)}

    Сейчас база в состоянии {html.escape(str(card.get("current_status", "-")))}, активных блокировок: {fmt_number(card.get("active_locks"))}, дней с последней активности: {fmt_number(card.get("days_since_last_activity"))}.

    Ключевые метрики

    {metric_cards}

    Карточка компании

    • Компания: {html.escape(str(card.get("company_name") or card.get("counterparty") or "-"))}
    • Нормализованное имя: {html.escape(str(card.get("normalized_counterparty") or "-"))}
    • Инфобаза: {html.escape(str(card.get("infobase") or "-"))}
    • Ответственный: {html.escape(str(card.get("registry_assignee_name") or card.get("owner_user") or "-"))}
    • Match mode: {html.escape(str(card.get("registry_match_mode") or "-"))}
    • ИНН / КПП: {html.escape(str(card.get("registry_inn") or "-"))} / {html.escape(str(card.get("registry_kpp") or "-"))}
    • Путь базы: {html.escape(str(card.get("base_path") or "-"))}

    Состояние файловой базы

    • Статус: {html.escape(str(company_state.get("current_status") or card.get("current_status") or "-"))}
    • Размер базы: {fmt_number(company_state.get("db_size_bytes") or card.get("db_size_bytes"))}
    • Размер reglog: {fmt_number(company_state.get("reglog_size_bytes") or card.get("reglog_size_bytes"))}
    • Активные блокировки: {fmt_number(company_state.get("active_locks") or card.get("active_locks"))}
    • Текущий activity score: {fmt_number(company_state.get("current_activity_score") or card.get("current_activity_score"))}
    • Последний snapshot: {html.escape(str(company_state.get("ts") or card.get("last_company_snapshot_at") or "-"))}

    Прогнозы

    {''.join(forecast_rows) or ''}
    Метрика Горизонт, д Baseline Прогноз total Confidence Примечание
    Нет данных.

    Сигналы

    {''.join(signal_rows) or ''}
    Время Severity Score Тип Комментарий
    Нет данных.

    Последние события

    {''.join(document_rows) or ''}
    Время Тип Операция Активность Статус Автор
    Нет данных.
    """ @app.get("/manager/company/{counterparty}", response_class=HTMLResponse) def manager_company_view(counterparty: str, infobase: str | None = None) -> str: return render_company_detail_html(company_summary(counterparty, infobase), infobase) if __name__ == "__main__": args = parse_args() uvicorn.run(app, host=args.host, port=args.port)