-
Раскладки по товарам (за 30 дн.)
-
+
+
Раскладки по товарам (за 30 дн.)
+
+
+
| Товар | Откуда | Куда |
A, тг/кг |
@@ -19,3 +24,13 @@
{{ ARB_HTML }}
+
+
diff --git a/src/export_excel.py b/src/export_excel.py
new file mode 100644
index 0000000..faafa5c
--- /dev/null
+++ b/src/export_excel.py
@@ -0,0 +1,149 @@
+"""CLI export: dump analytics to an .xlsx the user can grab from the Files tab.
+
+Uses the host `makexlsx` command (no openpyxl in the run container).
+Builds CSVs for each sheet, then shells out to makexlsx with a JSON of sheets.
+
+Usage:
+ python3 -m src.export_excel # latest 30d: dynamics + spreads + alerts
+ python3 -m src.export_excel --out data/report.xlsx
+ python3 -m src.export_excel --days 60
+"""
+from __future__ import annotations
+import argparse
+import csv
+import json
+import os
+import subprocess
+import tempfile
+from datetime import date, timedelta
+from pathlib import Path
+
+from . import db
+
+
+def _daily_series(product: str, days: int) -> list[dict]:
+ """Daily avg/min/max/aggregated rows for one product (mirrors dashboard series_for)."""
+ import statistics
+ to = date.today()
+ frm = to - timedelta(days=days)
+ rows = db.query_prices(product=product, from_date=frm, to_date=to,
+ include_quarantine=False, limit=10000)
+ agg: dict[str, list[float]] = {}
+ for r in rows:
+ if r.get("value_kg"):
+ agg.setdefault(r["as_of"], []).append(r["value_kg"])
+ out = []
+ for d in sorted(agg):
+ vals = agg[d]
+ out.append({"as_of": d, "n": len(vals),
+ "avg_val": round(statistics.mean(vals), 4),
+ "min_val": round(min(vals), 4), "max_val": round(max(vals), 4),
+ "med_val": round(statistics.median(vals), 4)})
+ return out
+
+
+def _sheet_dynamics(product: str, days: int) -> list[list[str]]:
+ """Daily series for one product: date, mean, min, max, n, delta%."""
+ series = _daily_series(product, days)
+ rows = [["Дата", "Среднее, тг/кг", "Мин", "Макс", "Записей", "Δ% к пред. дню"]]
+ prev = None
+ for s in series:
+ avg = s.get("avg_val")
+ pct = "" if prev is None or not avg else round((avg - prev) / prev * 100, 2)
+ rows.append([s["as_of"], round(avg or 0, 2),
+ round(s.get("min_val") or 0, 2), round(s.get("max_val") or 0, 2),
+ s.get("n", 0), pct])
+ if avg is not None:
+ prev = avg
+ return rows
+
+
+def _sheet_spreads(days: int) -> list[list[str]]:
+ today = date.today()
+ rows = db.query_prices(from_date=today - timedelta(days=days), to_date=today,
+ include_quarantine=False, limit=20000)
+ by_prod: dict[str, dict[str, list[dict]]] = {}
+ for r in rows:
+ by_prod.setdefault(r["product"], {}).setdefault(r["country"], []).append(r)
+ out = [["Товар", "Откуда", "Куда", "A, тг/кг", "B, тг/кг", "Δ, тг/кг", "Δ%", "Даты"]]
+ for p, cd in by_prod.items():
+ if len(cd) < 2:
+ continue
+ last_v: dict[str, tuple[float, str]] = {}
+ for c, rs in cd.items():
+ agg: dict[str, list[float]] = {}
+ for r in rs:
+ agg.setdefault(r["as_of"], []).append(r["value_kg"])
+ if not agg:
+ continue
+ last = max(agg)
+ last_v[c] = (sum(agg[last]) / len(agg[last]), last)
+ items = list(last_v.items()) # [(country, (val, date_str)), ...]
+ for i in range(len(items)):
+ for j in range(len(items)):
+ if i == j:
+ continue
+ ca, (va, da) = items[i]
+ cb, (vb, db2) = items[j]
+ if ca != "KZ" or cb == "KZ":
+ continue # only KZ → other pairs
+ delta = vb - va
+ pct = (delta / va * 100) if va else 0
+ out.append([p, ca, cb, round(va, 2), round(vb, 2), round(delta, 2),
+ round(pct, 1), f"{da[:10]} / {db2[:10]}"])
+ return out
+
+
+def _sheet_alerts(limit: int = 100) -> list[list[str]]:
+ rows = db.recent_alerts(limit=limit)
+ out = [["Вид", "Товар", "Регион", "Сообщение", "Серьёзность", "Создан"]]
+ for a in rows:
+ out.append([a.get("kind", ""), a.get("product", ""), a.get("region", ""),
+ a.get("message", ""), a.get("severity", ""),
+ (a.get("created_at") or "")[:16]])
+ return out
+
+
+def build_sheets(days: int) -> dict[str, list[list[str]]]:
+ products = [r["product"] for r in
+ db.exec_sql("SELECT DISTINCT product FROM prices WHERE quarantined=0 "
+ "ORDER BY product")][:8]
+ sheets: dict[str, list[list[str]]] = {}
+ for p in products:
+ short = p.replace("_", " ")[:31]
+ sheets[short[:31]] = _sheet_dynamics(p, days)
+ sheets["Раскладки"] = _sheet_spreads(days)
+ sheets["Алерты"] = _sheet_alerts(limit=100)
+ return sheets
+
+
+def to_xlsx(sheets: dict[str, list[list[str]]], out_path: str) -> bool:
+ """Write each sheet to a temp CSV, then call the host makexlsx with a JSON of sheets."""
+ payload = {name: rows for name, rows in sheets.items() if rows}
+ if not payload:
+ print("Нет данных для выгрузки.")
+ return False
+ proc = subprocess.run(["makexlsx", out_path], input=json.dumps(payload),
+ text=True, capture_output=True)
+ if proc.returncode != 0:
+ print("makexlsx error:", proc.stderr[:400])
+ return False
+ return True
+
+
+def main() -> None:
+ ap = argparse.ArgumentParser()
+ ap.add_argument("--out", default=str(Path("data") / "agromarket_report.xlsx"))
+ ap.add_argument("--days", type=int, default=30)
+ args = ap.parse_args()
+ os.makedirs(os.path.dirname(args.out) or ".", exist_ok=True)
+ sheets = build_sheets(args.days)
+ ok = to_xlsx(sheets, args.out)
+ if ok:
+ print(f"Готово: {args.out} | листов: {len(sheets)}")
+ for name in sheets:
+ print(f" - {name}: {len(sheets[name])} строк")
+
+
+if __name__ == "__main__":
+ main()