Spaces:
Running
Running
| """API theo dõi người dùng. | |
| Hai nhóm endpoint: | |
| - POST /api/events — app gửi log lên (không cần token). | |
| - GET /api/analytics/admin/* — quản trị đọc báo cáo (cần | |
| VIDEO_ADMIN_TOKEN, dùng chung mật khẩu với tab "Quản lý người dùng"). | |
| Vì app đăng nhập bằng tên (không có token phiên), endpoint nhận log KHÔNG xác | |
| thực được người gửi — đây là tool nội bộ nên chấp nhận, nhưng có chặn kích | |
| thước lô để không ai bơm được cả biển dữ liệu trong một lần gọi. | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import io | |
| import json as _json | |
| import time | |
| from datetime import datetime, timedelta, timezone | |
| from fastapi import APIRouter, HTTPException, Request | |
| from fastapi.responses import StreamingResponse | |
| from ..config import settings | |
| from ..services.analytics import etl | |
| from ..services.analytics import labels | |
| from ..services.analytics import storage as store | |
| from ..services.auth.storage import list_codes, list_users | |
| router = APIRouter(tags=["analytics"]) | |
| # Trần số event một lần POST (frontend gửi lô ~20). | |
| MAX_BATCH = 200 | |
| # Dọn log quá hạn nhiều nhất 1 lần/giờ, đi ké request có sẵn (khỏi cần cron). | |
| _last_purge = 0.0 | |
| _PURGE_EVERY = 3600.0 | |
| def _maybe_purge(): | |
| global _last_purge | |
| now = time.time() | |
| if now - _last_purge < _PURGE_EVERY: | |
| return | |
| _last_purge = now | |
| try: | |
| store.purge_old() | |
| except Exception as ex: # dọn dẹp lỗi thì kệ, không được làm hỏng request | |
| print(f"[analytics] purge error: {ex}") | |
| def _check(token: str): | |
| if not settings.VIDEO_ADMIN_TOKEN or token != settings.VIDEO_ADMIN_TOKEN: | |
| raise HTTPException(403, "Not authorized") | |
| def _tz(): | |
| return timezone(timedelta(hours=settings.ANALYTICS_TZ_OFFSET)) | |
| def _range(days: int, date_from: str, date_to: str) -> tuple[float, float]: | |
| """Đổi bộ lọc ngày (theo múi giờ báo cáo) thành khoảng epoch [from, to).""" | |
| tz = _tz() | |
| if date_from or date_to: | |
| try: | |
| start = (datetime.strptime(date_from, "%Y-%m-%d").replace(tzinfo=tz).timestamp() | |
| if date_from else 0.0) | |
| end = ((datetime.strptime(date_to, "%Y-%m-%d").replace(tzinfo=tz) + timedelta(days=1)).timestamp() | |
| if date_to else time.time() + 60) | |
| return start, end | |
| except ValueError: | |
| raise HTTPException(400, "Ngày phải theo dạng YYYY-MM-DD") | |
| days = max(1, min(int(days or 30), 365)) | |
| now = datetime.now(tz) | |
| start = (now - timedelta(days=days - 1)).replace(hour=0, minute=0, second=0, microsecond=0) | |
| return start.timestamp(), time.time() + 60 | |
| def _name_map() -> dict: | |
| """Mã U00N -> tên người dùng, để dashboard hiện TÊN thay vì chỉ mã. | |
| Log lưu user_id là mã (đã lowercase khi chuẩn hoá) nên key map cũng lowercase. | |
| """ | |
| out = {} | |
| try: | |
| for r in list_codes(): | |
| out[(r["code"] or "").lower()] = r["username"] | |
| except Exception: | |
| pass | |
| return out | |
| def _granted_accounts() -> list: | |
| """Danh sách tài khoản đã được duyệt (bảng phân quyền), lỗi -> rỗng.""" | |
| try: | |
| return list_users() | |
| except Exception: | |
| return [] | |
| # ── Nhận log ───────────────────────────────────────────────────── | |
| async def ingest(request: Request): | |
| """Nhận một lô event. Chấp cả `{"events":[...]}` lẫn mảng trần. | |
| `sendBeacon` lúc đóng tab gửi Content-Type lạ nên đọc body thủ công. | |
| """ | |
| try: | |
| body = await request.json() | |
| except Exception: | |
| return {"ok": False, "error": "Body không phải JSON", "saved": 0} | |
| items = body.get("events") if isinstance(body, dict) else body | |
| if not isinstance(items, list): | |
| return {"ok": False, "error": "Thiếu mảng events", "saved": 0} | |
| if len(items) > MAX_BATCH: | |
| items = items[:MAX_BATCH] | |
| now = time.time() | |
| rows = [store.normalize(it, now) for it in items if isinstance(it, dict)] | |
| # CHỈNH LỆCH ĐỒNG HỒ: dời cả lô theo đúng MỘT hằng số. Hiệu số giữa các dòng | |
| # (thời gian mỗi bước) giữ nguyên, nhưng mốc tuyệt đối khớp với dòng log do | |
| # backend ghi -> ghép chung một dòng thời gian được. | |
| # | |
| # LẤY GÌ LÀM MỐC ĐO LỆCH: | |
| # - Có `sent_at` (app desktop gửi kèm) -> giờ máy khách LÚC BẤM GỬI. App | |
| # có hàng đợi lưu đĩa: mất mạng thì gom lại, mở lần sau gửi bù. Nếu vẫn | |
| # lấy event mới nhất làm mốc thì cả lô cũ bị kéo về hiện tại và rơi sai | |
| # NGÀY (báo cáo gộp ngày theo server_ts). Dựa vào `sent_at` thì log cũ | |
| # giữ đúng tuổi của nó, mà đồng hồ máy sai vẫn được nắn. | |
| # - Không có (bản web, gửi mỗi 5 giây) -> giữ nguyên cách cũ: lấy event | |
| # mới nhất, sai số chỉ còn độ trễ mạng. | |
| sent_ts = store.parse_iso(str(body.get("sent_at") or "")) if isinstance(body, dict) else None | |
| if rows: | |
| base = sent_ts if sent_ts is not None else max(r["event_ts"] for r in rows) | |
| skew = now - base | |
| if abs(skew) > 1.0: | |
| for r in rows: | |
| r["event_ts"] += skew | |
| # GỬI BÙ: kéo cả `server_ts` về đúng lúc việc xảy ra. | |
| # | |
| # Nắn `event_ts` thôi là CHƯA ĐỦ. Báo cáo gộp "theo ngày" và lọc khoảng | |
| # ngày đều đọc `server_ts` (xem etl.local_date) — để nguyên là lúc nhận | |
| # thì việc làm tối qua bị đếm sang hôm nay, biểu đồ hoạt động hàng ngày | |
| # sai ngày dù thời lượng vẫn đúng. | |
| # | |
| # Chỉ làm với lô có `sent_at`, tức app desktop có hàng đợi lưu đĩa. Bản | |
| # web gửi mỗi 5 giây, không có `sent_at`, giữ nguyên hành vi cũ. | |
| # | |
| # An toàn: `event_ts` lúc này đã được neo theo giờ máy chủ nên tin được; | |
| # vẫn chặn hai đầu — không cho lùi quá hạn lưu trữ (dọn log sẽ xoá ngay | |
| # cái vừa ghi) và không cho vượt quá hiện tại. | |
| if sent_ts is not None: | |
| keep = settings.ANALYTICS_RETENTION_DAYS | |
| floor = (now - keep * 86400 + 3600) if keep and keep > 0 else 0.0 | |
| for r in rows: | |
| if r["event_ts"] < now - 60: # thật sự là log cũ | |
| r["server_ts"] = max(floor, min(r["event_ts"], now)) | |
| saved = store.insert(rows) | |
| _maybe_purge() | |
| return {"ok": True, "saved": saved, "received": len(rows)} | |
| # ── Báo cáo cho quản trị ───────────────────────────────────────── | |
| def admin_summary(token: str = "", days: int = 30, date_from: str = "", | |
| date_to: str = "", feature: str = "", user_id: str = ""): | |
| """Toàn bộ số liệu cho tab "Quản lý thời lượng".""" | |
| _check(token) | |
| start, end = _range(days, date_from, date_to) | |
| events = store.query(start, end, user_id=user_id, feature=feature) | |
| data = etl.summarize(events) | |
| data["range"] = {"from_ts": start, "to_ts": end, "days": days, | |
| "tz_offset": settings.ANALYTICS_TZ_OFFSET} | |
| data["storage"] = store.stats() | |
| data["user_names"] = _name_map() | |
| # Số tài khoản ĐƯỢC CẤP (khác với số người thật sự vào dùng) — đếm bảng | |
| # tài khoản đã duyệt, đúng bằng số dòng tab "Quản lý người dùng". | |
| # | |
| # KHÔNG đếm bảng mã U00N: mã chỉ sinh khi người đó đăng nhập lần đầu (hoặc | |
| # khi quản trị tự tay thêm), còn duyệt từ hàng chờ thì không cấp mã — nên | |
| # đếm mã ra số người đã từng vào, ít hơn hẳn số tài khoản được cấp. | |
| data["granted_accounts"] = len(_granted_accounts()) | |
| return {"ok": True, **data} | |
| def admin_user_detail(token: str = "", user_id: str = "", days: int = 30, | |
| date_from: str = "", date_to: str = ""): | |
| """Chi tiết một người: từng phiên -> từng task -> từng bước.""" | |
| _check(token) | |
| if not user_id.strip(): | |
| raise HTTPException(400, "Thiếu user_id") | |
| start, end = _range(days, date_from, date_to) | |
| events = store.query(start, end, user_id=user_id) | |
| detail = etl.user_detail(events, user_id.strip().lower()) | |
| detail["name"] = _name_map().get(user_id.strip().lower(), "") | |
| return {"ok": True, **detail} | |
| def admin_raw_events(token: str = "", days: int = 7, date_from: str = "", | |
| date_to: str = "", user_id: str = "", feature: str = "", | |
| task_id: str = "", limit: int = 500): | |
| """Log thô — để soi đúng dòng khi số liệu trông lạ. | |
| Kèm sẵn bản tiếng Việt của từng dòng (`vi_*`) để màn hình chỉ việc in ra: | |
| bảng dịch nằm một chỗ ở backend, khỏi chép đôi sang frontend rồi lệch nhau. | |
| """ | |
| _check(token) | |
| start, end = _range(days, date_from, date_to) | |
| rows = store.query(start, end, user_id=user_id, feature=feature, | |
| task_id=task_id, limit=max(1, min(int(limit or 500), 5000)), | |
| newest_first=True) | |
| names = _name_map() | |
| for r in rows: | |
| uid = r.get("user_id", "") | |
| r["vi_user"] = names.get(uid, uid) | |
| r["vi_feature"] = labels.feature_label(r.get("feature", "")) | |
| r["vi_step"] = labels.step_label(r.get("step", "")) | |
| r["vi_event"] = labels.event_label(r.get("event_name", "")) | |
| r["vi_value"] = labels.value_label(r.get("event_value", "")) | |
| r["vi_latency"] = labels.duration_label(r.get("latency_ms")) | |
| r["vi_detail"] = labels.describe_props(r.get("properties") or {}) | |
| return {"ok": True, "events": rows, "count": len(rows)} | |
| def admin_export_csv(token: str = "", days: int = 30, date_from: str = "", | |
| date_to: str = "", user_id: str = "", feature: str = "", | |
| raw: int = 0): | |
| """Xuất log ra CSV. | |
| Mặc định xuất bản ĐỌC ĐƯỢC: mỗi cột là một câu tiếng Việt (tên người, tên | |
| tool, tên bước, tên hành động, thời gian máy chạy), cục `properties_json` | |
| được bung thành "Bước máy: Bóc lời & dịch · Lần bấm chạy thứ: 1". Ba cột mã | |
| (`Mã việc`/`Mã phiên`/`Mã dòng`) và cột JSON gốc vẫn nằm ở cuối để còn dò | |
| ngược được khi số liệu trông lạ. | |
| `?raw=1` -> bản 11 cột đúng `raw_events_schema_v2.xlsx` (cho bên nào cần | |
| nạp thẳng vào công cụ khác, đừng đổi thứ tự cột của nó). | |
| """ | |
| _check(token) | |
| start, end = _range(days, date_from, date_to) | |
| rows = store.query(start, end, user_id=user_id, feature=feature) | |
| names = _name_map() | |
| buf = io.StringIO() | |
| w = csv.writer(buf) | |
| if raw: | |
| w.writerow(["event_id", "event_time", "user_id", "session_id", "task_id", | |
| "feature", "step", "event_name", "event_value", "latency_ms", | |
| "properties_json"]) | |
| for r in rows: | |
| w.writerow([ | |
| r.get("event_id", ""), r.get("event_time", ""), r.get("user_id", ""), | |
| r.get("session_id", ""), r.get("task_id", ""), r.get("feature", ""), | |
| r.get("step", ""), r.get("event_name", ""), r.get("event_value", ""), | |
| "" if r.get("latency_ms") is None else r["latency_ms"], | |
| _json.dumps(r.get("properties") or {}, ensure_ascii=False), | |
| ]) | |
| else: | |
| w.writerow(["STT", "Ngày", "Giờ", "Người dùng", "Mã người dùng", "Tool", | |
| "Bước", "Hành động", "Giá trị", "Máy chạy", "Chi tiết", | |
| "Mã việc", "Mã phiên", "Mã dòng", "Thời gian gốc (ISO)", | |
| "Chi tiết gốc (JSON)"]) | |
| tz = _tz() | |
| for i, r in enumerate(rows, 1): | |
| uid = r.get("user_id", "") | |
| dt = datetime.fromtimestamp(r.get("event_ts") or 0.0, tz) | |
| props = r.get("properties") or {} | |
| w.writerow([ | |
| i, | |
| dt.strftime("%d/%m/%Y"), | |
| dt.strftime("%H:%M:%S"), | |
| names.get(uid, uid), | |
| uid.upper(), | |
| labels.feature_label(r.get("feature", "")), | |
| labels.step_label(r.get("step", "")), | |
| labels.event_label(r.get("event_name", "")), | |
| labels.value_label(r.get("event_value", "")), | |
| labels.duration_label(r.get("latency_ms")), | |
| labels.describe_props(props), | |
| r.get("task_id", ""), | |
| r.get("session_id", ""), | |
| r.get("event_id", ""), | |
| r.get("event_time", ""), | |
| _json.dumps(props, ensure_ascii=False) if props else "", | |
| ]) | |
| buf.seek(0) | |
| kind = "raw" if raw else "doc_duoc" | |
| fname = f"nhat_ky_su_dung_{kind}_{datetime.now(_tz()).strftime('%Y%m%d')}.csv" | |
| return StreamingResponse( | |
| # BOM để Excel mở tiếng Việt không lỗi phông | |
| iter(["" + buf.getvalue()]), | |
| media_type="text/csv; charset=utf-8", | |
| headers={"Content-Disposition": f'attachment; filename="{fname}"'}, | |
| ) | |
| def admin_storage(token: str = ""): | |
| _check(token) | |
| return {"ok": True, **store.stats()} | |
| def admin_purge(body: dict): | |
| """Dọn tay log cũ hơn `days` ngày.""" | |
| _check(body.get("token") or "") | |
| days = body.get("days") | |
| deleted = store.purge_old(int(days) if days else None) | |
| return {"ok": True, "deleted": deleted} | |