#!/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 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 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"
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)}
JSON
Markdown
History
Grafana
Краткий комментарий
{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 '
Нет данных.
'}
Что проверить руководителю
Ограничения интерпретации
Свежесть источников
Источник
Последняя отметка
Отставание, ч
Статус
{''.join(freshness_rows)}
"""
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]:
history_dir = manager_brief_state_dir() / "history"
if not history_dir.exists():
return {"items": [], "count": 0}
items: list[dict[str, Any]] = []
for path in sorted(history_dir.glob("*.json"), reverse=True)[:limit]:
payload = json.loads(path.read_text(encoding="utf-8"))
items.append(
{
"generated_at": payload.get("generated_at"),
"render_mode": payload.get("render_mode"),
"model": payload.get("model"),
"headline": payload.get("brief", {}).get("headline", ""),
"path": path.name,
}
)
return {"items": items, "count": len(items)}
@app.get("/manager/brief", response_class=HTMLResponse)
def manager_brief_view() -> str:
return render_manager_brief_html(load_latest_manager_brief())
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)}
"
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
Управленческий комментарий
{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}
Карточка компании
Состояние файловой базы
Прогнозы
Метрика
Горизонт, д
Baseline
Прогноз total
Confidence
Примечание
{''.join(forecast_rows) or 'Нет данных.
Сигналы
Время
Severity
Score
Тип
Комментарий
{''.join(signal_rows) or 'Нет данных.
Последние события
Время
Тип
Операция
Активность
Статус
Автор
{''.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)