refactor: chia nốt theme.py, i18n.py, usage_tracker.py — Gamma hết file vượt 400 dòng
Ba file dữ liệu cuối cùng của Gamma còn trên ngưỡng CASAN Check 2.
i18n.py 3075 -> 94
Dict STRINGS 3.000 dòng cắt thành 10 cụm theo đúng mốc phân đoạn có sẵn
trong file (mỗi mốc là một màn/hộp thoại), cụm nào quá dài thì cắt tiếp ở
ranh giới khoá. i18n.py giờ chỉ gộp lại và giữ 4 hàm set_language/
get_language/tr/on_language_changed.
Kiểm bằng cách so với bản gốc lấy từ git: 1437 mục / 1431 khoá duy nhất
(bản gốc vốn có 6 khoá lặp), sau khi chia vẫn 1431, KHÔNG thiếu khoá nào,
KHÔNG thừa khoá nào, KHÔNG giá trị nào lệch. Thứ tự gộp giữ nguyên nên
quy tắc "khoá trùng thì bản sau thắng" không đổi.
theme.py 907 -> 130
theme_palettes.py 328 hai bảng màu Tối/Sáng + lớp Palette
theme_qss.py 198 nửa vỏ (reset + shell)
theme_qss_controls.py 307 nửa điều khiển (nút, ô nhập, tab, badge)
Khuôn QSS 470 dòng cắt đôi đúng mốc `/* ---- surfaces */` của chính nó.
Đã đối chiếu: stylesheet('dark') ra đúng 24762 ký tự y như trước — khớp
từng byte, không phải "trông có vẻ giống".
core/usage_tracker.py 536 -> 307
usage_cost.py 101 bảng giá, quy đổi token sang tiền, định dạng
usage_periods.py 144 gộp theo ngày/tuần/tháng/quý, chuỗi vẽ biểu đồ
usage_ai_report.py 56 dựng câu nhắc cho AI phân tích
HAI LẦN TỰ CẮT HỎNG, ĐỀU CÙNG MỘT GỐC
--------------------------------------
1. Cắt theo m.lineno mà quên dòng @decorator phía trên -> @dataclass của
Palette bị bỏ lại mồ côi, "Palette() takes no arguments".
2. Đọc số dòng từ AST GỐC trong khi danh sách dòng đã bị cắt -> lần bóc thứ
hai dùng toạ độ cũ và cắt vào giữa một chữ ký hàm.
Cả hai lộ ngay vì mỗi script tự parse lại sau khi ghi. Bài học đã áp vào cả
ba lần chia: parse lại sau mỗi lần cắt, và luôn tính cả decorator.
KẾT QUẢ CASAN CHECK 2
---------------------
Nam 0 file vượt 400 (trước: 4, tổng 5.898 dòng)
Hiệp 0 (trước: 1)
Lâm 0 (trước: 1)
file mới 0 (61 file dưới presentation/ application/
domain/ infrastructure/ — chưa cái nào vượt)
Gamma sạch. 23 file còn vượt đều thuộc team khác (chat_panel.py 1802,
folder_tab.py 1589, structure_graph_view.py 1034...) — cần báo lên sớm chứ
đừng để tới hạn 30/08 mới lộ.
714 test xanh. 24/24 checker qua. CASAN Check 1 sạch.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
Nam Pham Dinh Thanh
co-authored by
Claude Opus 5
Claude Opus 5
parent
0e00bf3c2f
commit
6c68417103
@@ -0,0 +1,56 @@
|
||||
"""Dựng câu nhắc cho AI phân tích mức dùng — R09-T02.
|
||||
|
||||
Chỉ sinh văn bản. Tách riêng vì đây là phần dễ đổi nhất (câu chữ, cột hiển
|
||||
thị) và không liên quan tới việc ghi nhận hay tính tiền.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from ..config import CONFIG_DIR
|
||||
from . import model_pricing as mp
|
||||
from .usage_periods import period_breakdown, period_range_label
|
||||
|
||||
_AI_ANALYSIS_HEADERS = {
|
||||
"vi": ("Nhận xét thói quen", "Cách viết prompt tiết kiệm hơn", "Hành động giảm token"),
|
||||
"en": ("Usage habits", "Writing more efficient prompts", "Actions to cut token usage"),
|
||||
"ja": ("利用傾向", "より効率的なプロンプトの書き方", "トークン削減のためのアクション"),
|
||||
}
|
||||
|
||||
def build_ai_analysis_prompt(summary: Dict[str, Any], language: str = "vi") -> str:
|
||||
"""The prompt sent to the model for '✨ AI analyze my usage': aggregated
|
||||
numbers only — never raw prompt contents — asking for concrete habits
|
||||
feedback and token-saving recommendations, in the CURRENTLY SELECTED
|
||||
display language (headers included — not just the model's free-text reply,
|
||||
which would otherwise leave the section titles in Vietnamese regardless of
|
||||
the app's language setting)."""
|
||||
lang_names = {"vi": "Vietnamese", "ja": "Japanese", "en": "English"}
|
||||
h1, h2, h3 = _AI_ANALYSIS_HEADERS.get(language, _AI_ANALYSIS_HEADERS["vi"])
|
||||
top = "\n".join(f"- {label}: {tok:,} tokens"
|
||||
for label, tok in summary.get("top_labels", []))
|
||||
by_source = ", ".join(f"{k}={v:,}" for k, v in summary.get("by_source", []))
|
||||
return (
|
||||
"You are a token-efficiency coach for an AI desktop app (chat tabs + "
|
||||
"scheduled agent tasks). Analyze this usage summary and give the user "
|
||||
"practical advice, replying in "
|
||||
f"{lang_names.get(language, 'Vietnamese')}.\n\n"
|
||||
f"Period stats: {summary.get('turns', 0)} turns, "
|
||||
f"input={summary.get('in', 0):,} tokens, output={summary.get('out', 0):,}, "
|
||||
f"cache={summary.get('cache', 0):,}, "
|
||||
f"avg per prompt={summary.get('avg_per_turn', 0):,}.\n"
|
||||
f"Top consumers:\n{top or '- (none)'}\n"
|
||||
f"By area: {by_source or '(none)'}\n"
|
||||
f"Busiest day: {summary.get('busiest_day')} · busiest hour: {summary.get('busiest_hour')}\n\n"
|
||||
"Reply with EXACTLY these 3 short sections, in markdown, using THESE "
|
||||
f"section headers verbatim (already in {lang_names.get(language, 'Vietnamese')}):\n"
|
||||
f"1. **{h1}** — 2-3 bullet points about the usage pattern.\n"
|
||||
f"2. **{h2}** — 3 concrete prompt-writing tips "
|
||||
"tailored to the numbers above (e.g. long inputs → attach less / summarize "
|
||||
"first; many small turns → batch questions).\n"
|
||||
f"3. **{h3}** — 2-3 app-level actions (compact history, "
|
||||
"smaller model for simple tasks, reuse task outputs instead of re-asking).\n"
|
||||
"Keep the whole reply under 250 words."
|
||||
)
|
||||
@@ -0,0 +1,101 @@
|
||||
"""Bảng giá và quy đổi token thành tiền — R09-T02.
|
||||
|
||||
Tách khỏi ``usage_tracker.py``: ghi nhận mức dùng và tính tiền là hai việc
|
||||
khác nhau. Bảng giá đổi theo nhà cung cấp, cách ghi nhận thì không.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from ..config import CONFIG_DIR
|
||||
from . import model_pricing as mp
|
||||
|
||||
DEFAULT_PRICING = {
|
||||
"price_per_mtok_in_usd": 0.5, # USD per 1M input tokens (flat fallback rate)
|
||||
"price_per_mtok_out_usd": 1.5, # USD per 1M output tokens
|
||||
"price_per_mtok_cache_usd": 0.1, # USD per 1M cached tokens
|
||||
"currency": "USD", # display currency: USD | VND | JPY
|
||||
"usd_to_vnd": 25000.0,
|
||||
"usd_to_jpy": 150.0,
|
||||
# Per-model price table (USD / 1M tokens): {model: {"in","out","cache"}}.
|
||||
# Events whose model has an entry are costed with ITS rates; everything
|
||||
# else falls back to the flat price_per_mtok_* rates above. Edited in the
|
||||
# Monitoring Overview's pricing table.
|
||||
"model_prices": {},
|
||||
# Reference URL of the price list the table was filled from (set in
|
||||
# Settings; shown as a link beside the table — informational only, the
|
||||
# app never scrapes it).
|
||||
"pricing_url": "",
|
||||
}
|
||||
|
||||
_CURRENCY_FMT = {"USD": ("$", 4), "VND": ("₫", 0), "JPY": ("¥", 1)}
|
||||
|
||||
SUPPORTED_CURRENCIES = tuple(_CURRENCY_FMT)
|
||||
|
||||
def cost_usd(summary: Dict[str, Any], pricing: Dict[str, Any]) -> Dict[str, float]:
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
return {
|
||||
"in": summary.get("in", 0) / 1e6 * float(p["price_per_mtok_in_usd"]),
|
||||
"out": summary.get("out", 0) / 1e6 * float(p["price_per_mtok_out_usd"]),
|
||||
"cache": summary.get("cache", 0) / 1e6 * float(p["price_per_mtok_cache_usd"]),
|
||||
}
|
||||
|
||||
def cost_usd_events(events: List[Dict[str, Any]], pricing: Dict[str, Any]) -> Dict[str, float]:
|
||||
"""Per-bucket USD cost computed EVENT BY EVENT so the per-model price
|
||||
table applies: an event whose ``model`` has an entry in
|
||||
``pricing["model_prices"]`` is costed with that model's own rates; any
|
||||
other event uses the flat ``price_per_mtok_*`` rates. With an empty
|
||||
table this equals ``cost_usd(summarize(events), pricing)`` exactly."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
table = p.get("model_prices") or {}
|
||||
flat = {"in": float(p["price_per_mtok_in_usd"]),
|
||||
"out": float(p["price_per_mtok_out_usd"]),
|
||||
"cache": float(p["price_per_mtok_cache_usd"])}
|
||||
out = {"in": 0.0, "out": 0.0, "cache": 0.0}
|
||||
for e in events:
|
||||
rates = table.get(e.get("model", "")) or {}
|
||||
for bucket in ("in", "out", "cache"):
|
||||
try:
|
||||
rate = float(rates.get(bucket, flat[bucket]))
|
||||
except (TypeError, ValueError):
|
||||
rate = flat[bucket]
|
||||
out[bucket] += e.get(bucket, 0) / 1e6 * rate
|
||||
return out
|
||||
|
||||
def format_cost(usd: float, pricing: Dict[str, Any], digits: Optional[int] = None) -> str:
|
||||
"""Format a USD amount in the display currency. ``digits`` caps the number
|
||||
of decimal places (e.g. ``digits=2`` for the Total cost / Budget cards, so
|
||||
USD shows $1.23 not the default up-to-4 $1.2345) — never ADDS decimals to a
|
||||
currency that uses fewer (VND stays whole, JPY one place)."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
cur = p.get("currency", "USD")
|
||||
rate = {"USD": 1.0, "VND": float(p["usd_to_vnd"]), "JPY": float(p["usd_to_jpy"])}.get(cur, 1.0)
|
||||
symbol, cur_digits = _CURRENCY_FMT.get(cur, ("$", 2))
|
||||
if digits is not None:
|
||||
cur_digits = min(cur_digits, digits)
|
||||
value = usd * rate
|
||||
return f"{symbol}{value:,.{cur_digits}f}"
|
||||
|
||||
def format_cost_compact(usd: float, pricing: Dict[str, Any]) -> str:
|
||||
"""Compact cost format for the Dashboard chart's y-axis/endpoint labels —
|
||||
always 2 decimals (not format_cost's up-to-4 for USD) and abbreviated with
|
||||
K/M above 1,000/1,000,000, same convention as ``fmt_tokens``. The chart's
|
||||
y-axis label box is narrow; the longer full-precision string used to
|
||||
overflow it, visually clipping/obscuring the leading currency symbol."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
cur = p.get("currency", "USD")
|
||||
rate = {"USD": 1.0, "VND": float(p["usd_to_vnd"]), "JPY": float(p["usd_to_jpy"])}.get(cur, 1.0)
|
||||
symbol, _digits = _CURRENCY_FMT.get(cur, ("$", 2))
|
||||
value = usd * rate
|
||||
sign = "-" if value < 0 else ""
|
||||
value = abs(value)
|
||||
if value >= 1_000_000:
|
||||
body = f"{value / 1_000_000:,.2f}M"
|
||||
elif value >= 1_000:
|
||||
body = f"{value / 1_000:,.2f}K"
|
||||
else:
|
||||
body = f"{value:,.2f}"
|
||||
return f"{sign}{symbol}{body}"
|
||||
@@ -0,0 +1,144 @@
|
||||
"""Gộp mức dùng theo khoảng thời gian — R09-T02.
|
||||
|
||||
Ngày / tuần / tháng / quý: ranh giới khoảng, nhãn hiển thị, chuỗi số vẽ biểu
|
||||
đồ. Thuần tính toán trên danh sách sự kiện, không đụng đĩa.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import threading
|
||||
from datetime import date, datetime
|
||||
from pathlib import Path
|
||||
from typing import Any, Dict, List, Optional
|
||||
from ..config import CONFIG_DIR
|
||||
from . import model_pricing as mp
|
||||
from .usage_cost import cost_usd_events
|
||||
|
||||
def bucketed_series(events: List[Dict[str, Any]], granularity: str = "day",
|
||||
pricing: Dict[str, Any] = None, last: int = None) -> List[tuple]:
|
||||
"""Group usage events into time buckets → ordered ``[(label, tokens, cost_usd)]``.
|
||||
|
||||
``granularity``: ``day`` (YYYY-MM-DD) · ``month`` (YYYY-MM) · ``year`` (YYYY).
|
||||
``last`` keeps only the most recent N buckets (for the dashboard chart)."""
|
||||
from collections import OrderedDict
|
||||
pricing = pricing or {}
|
||||
|
||||
def _key(ts: Any) -> str:
|
||||
s = str(ts or "")[:10]
|
||||
if granularity == "year":
|
||||
return s[:4]
|
||||
if granularity == "month":
|
||||
return s[:7]
|
||||
return s
|
||||
|
||||
buckets: "OrderedDict[str, List[Dict[str, Any]]]" = OrderedDict()
|
||||
for e in sorted(events, key=lambda ev: str(ev.get("ts", ""))):
|
||||
k = _key(e.get("ts"))
|
||||
if k:
|
||||
buckets.setdefault(k, []).append(e)
|
||||
out = []
|
||||
for k, evs in buckets.items():
|
||||
tokens = sum(int(e.get("in", 0) or 0) + int(e.get("out", 0) or 0)
|
||||
+ int(e.get("cache", 0) or 0) for e in evs)
|
||||
cost = sum(cost_usd_events(evs, pricing).values())
|
||||
out.append((k, tokens, cost))
|
||||
if last and len(out) > last:
|
||||
out = out[-last:]
|
||||
return out
|
||||
|
||||
def period_bounds(gran: str, offset: int, today: Optional[date] = None) -> tuple:
|
||||
"""[start, end) dates of the period ``offset`` periods from the current one
|
||||
(0 = current, -1 = the previous week/month/year). Weeks run Mon→Sun."""
|
||||
from datetime import timedelta
|
||||
today = today or date.today()
|
||||
if gran == "week":
|
||||
monday = today - timedelta(days=today.weekday()) # Monday of this week
|
||||
start = monday + timedelta(weeks=offset)
|
||||
return start, start + timedelta(days=7)
|
||||
if gran == "year":
|
||||
y = today.year + offset
|
||||
return date(y, 1, 1), date(y + 1, 1, 1)
|
||||
# month (default)
|
||||
base = today.year * 12 + (today.month - 1) + offset
|
||||
y, m = divmod(base, 12)
|
||||
y2, m2 = divmod(base + 1, 12)
|
||||
return date(y, m + 1, 1), date(y2, m2 + 1, 1)
|
||||
|
||||
def _period_label(gran: str, start: date) -> str:
|
||||
if gran == "week":
|
||||
return start.isoformat() # the week's Monday (YYYY-MM-DD)
|
||||
if gran == "year":
|
||||
return str(start.year)
|
||||
return start.strftime("%Y-%m")
|
||||
|
||||
def _sum_between(events: List[Dict[str, Any]], start: date, end: date,
|
||||
pricing: Dict[str, Any]) -> tuple:
|
||||
lo, hi = start.isoformat(), end.isoformat()
|
||||
evs = [e for e in events if lo <= str(e.get("ts", ""))[:10] < hi]
|
||||
tokens = sum(int(e.get("in", 0) or 0) + int(e.get("out", 0) or 0)
|
||||
+ int(e.get("cache", 0) or 0) for e in evs)
|
||||
cost = sum(cost_usd_events(evs, pricing).values()) if evs else 0.0
|
||||
return tokens, cost
|
||||
|
||||
def period_totals(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
offset: int = 0, today: Optional[date] = None) -> tuple:
|
||||
"""(tokens, cost_usd) for the single period ``offset`` periods from now."""
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
return _sum_between(events, start, end, pricing)
|
||||
|
||||
def period_window(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
count: int, offset: int = 0, today: Optional[date] = None) -> List[tuple]:
|
||||
"""``count`` consecutive, ZERO-FILLED periods ending at (current + offset),
|
||||
ordered oldest→newest → ``[(label, tokens, cost_usd)]``. ``offset`` (≤ 0)
|
||||
pages the window into the past for the Dashboard's prev/next navigation."""
|
||||
out = []
|
||||
for i in range(count - 1, -1, -1):
|
||||
start, end = period_bounds(gran, offset - i, today)
|
||||
tok, cost = _sum_between(events, start, end, pricing)
|
||||
out.append((_period_label(gran, start), tok, cost))
|
||||
return out
|
||||
|
||||
def period_breakdown(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
offset: int = 0, today: Optional[date] = None) -> List[tuple]:
|
||||
"""Break the SELECTED period (``offset`` periods from now) into its sub-parts
|
||||
→ ``[(label, tokens, cost_usd)]``:
|
||||
· week → 7 days Mon→Sun (label ``MM/DD``)
|
||||
· month → weeks W1…Wn (7-day chunks from the 1st)
|
||||
· year → 12 months (label ``01``…``12``)."""
|
||||
from datetime import timedelta
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
out = []
|
||||
if gran == "week":
|
||||
for i in range(7):
|
||||
d = start + timedelta(days=i)
|
||||
tok, cost = _sum_between(events, d, d + timedelta(days=1), pricing)
|
||||
out.append((d.strftime("%m/%d"), tok, cost))
|
||||
elif gran == "year":
|
||||
for m in range(1, 13):
|
||||
ms = date(start.year, m, 1)
|
||||
me = date(start.year + 1, 1, 1) if m == 12 else date(start.year, m + 1, 1)
|
||||
tok, cost = _sum_between(events, ms, me, pricing)
|
||||
out.append((f"{m:02d}", tok, cost))
|
||||
else: # month → weeks W1..Wn
|
||||
ndays = (end - start).days
|
||||
wk, day = 1, 1
|
||||
while day <= ndays:
|
||||
ws = date(start.year, start.month, day)
|
||||
we = date(start.year, start.month, day + 7) if day + 7 <= ndays else end
|
||||
tok, cost = _sum_between(events, ws, we, pricing)
|
||||
out.append((f"W{wk}", tok, cost))
|
||||
wk += 1
|
||||
day += 7
|
||||
return out
|
||||
|
||||
def period_range_label(gran: str, offset: int, today: Optional[date] = None) -> str:
|
||||
"""Human label for the selected period (shown in the Dashboard header) —
|
||||
week → MM/DD – MM/DD, month → YYYY/MM, year → YYYY."""
|
||||
from datetime import timedelta
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
if gran == "week":
|
||||
last_day = end - timedelta(days=1)
|
||||
return f"{start.strftime('%m/%d')} – {last_day.strftime('%m/%d')}"
|
||||
if gran == "year":
|
||||
return str(start.year)
|
||||
return start.strftime("%Y/%m")
|
||||
+11
-240
@@ -12,6 +12,17 @@ The turn's source/label is set by the caller ON THE WORKER THREAD via
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
# Giữ đường vào cũ: nhiều nơi import mấy tên này thẳng từ usage_tracker.
|
||||
from .usage_ai_report import build_ai_analysis_prompt # noqa: F401
|
||||
from .usage_cost import ( # noqa: F401
|
||||
DEFAULT_PRICING, SUPPORTED_CURRENCIES, cost_usd, cost_usd_events,
|
||||
format_cost, format_cost_compact,
|
||||
)
|
||||
from .usage_periods import ( # noqa: F401
|
||||
bucketed_series, period_bounds, period_breakdown, period_range_label,
|
||||
period_totals, period_window,
|
||||
)
|
||||
|
||||
import json
|
||||
import threading
|
||||
from datetime import date, datetime
|
||||
@@ -204,198 +215,30 @@ def summarize(events: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||||
|
||||
|
||||
# ---- cost ------------------------------------------------------------------
|
||||
DEFAULT_PRICING = {
|
||||
"price_per_mtok_in_usd": 0.5, # USD per 1M input tokens (flat fallback rate)
|
||||
"price_per_mtok_out_usd": 1.5, # USD per 1M output tokens
|
||||
"price_per_mtok_cache_usd": 0.1, # USD per 1M cached tokens
|
||||
"currency": "USD", # display currency: USD | VND | JPY
|
||||
"usd_to_vnd": 25000.0,
|
||||
"usd_to_jpy": 150.0,
|
||||
# Per-model price table (USD / 1M tokens): {model: {"in","out","cache"}}.
|
||||
# Events whose model has an entry are costed with ITS rates; everything
|
||||
# else falls back to the flat price_per_mtok_* rates above. Edited in the
|
||||
# Monitoring Overview's pricing table.
|
||||
"model_prices": {},
|
||||
# Reference URL of the price list the table was filled from (set in
|
||||
# Settings; shown as a link beside the table — informational only, the
|
||||
# app never scrapes it).
|
||||
"pricing_url": "",
|
||||
}
|
||||
|
||||
_CURRENCY_FMT = {"USD": ("$", 4), "VND": ("₫", 0), "JPY": ("¥", 1)}
|
||||
|
||||
# Currencies the display picker offers — exactly the ones format_cost() can
|
||||
# actually convert to (symbol/precision above + a usd_to_* rate below).
|
||||
SUPPORTED_CURRENCIES = tuple(_CURRENCY_FMT)
|
||||
|
||||
|
||||
def cost_usd(summary: Dict[str, Any], pricing: Dict[str, Any]) -> Dict[str, float]:
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
return {
|
||||
"in": summary.get("in", 0) / 1e6 * float(p["price_per_mtok_in_usd"]),
|
||||
"out": summary.get("out", 0) / 1e6 * float(p["price_per_mtok_out_usd"]),
|
||||
"cache": summary.get("cache", 0) / 1e6 * float(p["price_per_mtok_cache_usd"]),
|
||||
}
|
||||
|
||||
|
||||
def cost_usd_events(events: List[Dict[str, Any]], pricing: Dict[str, Any]) -> Dict[str, float]:
|
||||
"""Per-bucket USD cost computed EVENT BY EVENT so the per-model price
|
||||
table applies: an event whose ``model`` has an entry in
|
||||
``pricing["model_prices"]`` is costed with that model's own rates; any
|
||||
other event uses the flat ``price_per_mtok_*`` rates. With an empty
|
||||
table this equals ``cost_usd(summarize(events), pricing)`` exactly."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
table = p.get("model_prices") or {}
|
||||
flat = {"in": float(p["price_per_mtok_in_usd"]),
|
||||
"out": float(p["price_per_mtok_out_usd"]),
|
||||
"cache": float(p["price_per_mtok_cache_usd"])}
|
||||
out = {"in": 0.0, "out": 0.0, "cache": 0.0}
|
||||
for e in events:
|
||||
rates = table.get(e.get("model", "")) or {}
|
||||
for bucket in ("in", "out", "cache"):
|
||||
try:
|
||||
rate = float(rates.get(bucket, flat[bucket]))
|
||||
except (TypeError, ValueError):
|
||||
rate = flat[bucket]
|
||||
out[bucket] += e.get(bucket, 0) / 1e6 * rate
|
||||
return out
|
||||
|
||||
|
||||
def bucketed_series(events: List[Dict[str, Any]], granularity: str = "day",
|
||||
pricing: Dict[str, Any] = None, last: int = None) -> List[tuple]:
|
||||
"""Group usage events into time buckets → ordered ``[(label, tokens, cost_usd)]``.
|
||||
|
||||
``granularity``: ``day`` (YYYY-MM-DD) · ``month`` (YYYY-MM) · ``year`` (YYYY).
|
||||
``last`` keeps only the most recent N buckets (for the dashboard chart)."""
|
||||
from collections import OrderedDict
|
||||
pricing = pricing or {}
|
||||
|
||||
def _key(ts: Any) -> str:
|
||||
s = str(ts or "")[:10]
|
||||
if granularity == "year":
|
||||
return s[:4]
|
||||
if granularity == "month":
|
||||
return s[:7]
|
||||
return s
|
||||
|
||||
buckets: "OrderedDict[str, List[Dict[str, Any]]]" = OrderedDict()
|
||||
for e in sorted(events, key=lambda ev: str(ev.get("ts", ""))):
|
||||
k = _key(e.get("ts"))
|
||||
if k:
|
||||
buckets.setdefault(k, []).append(e)
|
||||
out = []
|
||||
for k, evs in buckets.items():
|
||||
tokens = sum(int(e.get("in", 0) or 0) + int(e.get("out", 0) or 0)
|
||||
+ int(e.get("cache", 0) or 0) for e in evs)
|
||||
cost = sum(cost_usd_events(evs, pricing).values())
|
||||
out.append((k, tokens, cost))
|
||||
if last and len(out) > last:
|
||||
out = out[-last:]
|
||||
return out
|
||||
|
||||
|
||||
def period_bounds(gran: str, offset: int, today: Optional[date] = None) -> tuple:
|
||||
"""[start, end) dates of the period ``offset`` periods from the current one
|
||||
(0 = current, -1 = the previous week/month/year). Weeks run Mon→Sun."""
|
||||
from datetime import timedelta
|
||||
today = today or date.today()
|
||||
if gran == "week":
|
||||
monday = today - timedelta(days=today.weekday()) # Monday of this week
|
||||
start = monday + timedelta(weeks=offset)
|
||||
return start, start + timedelta(days=7)
|
||||
if gran == "year":
|
||||
y = today.year + offset
|
||||
return date(y, 1, 1), date(y + 1, 1, 1)
|
||||
# month (default)
|
||||
base = today.year * 12 + (today.month - 1) + offset
|
||||
y, m = divmod(base, 12)
|
||||
y2, m2 = divmod(base + 1, 12)
|
||||
return date(y, m + 1, 1), date(y2, m2 + 1, 1)
|
||||
|
||||
|
||||
def _period_label(gran: str, start: date) -> str:
|
||||
if gran == "week":
|
||||
return start.isoformat() # the week's Monday (YYYY-MM-DD)
|
||||
if gran == "year":
|
||||
return str(start.year)
|
||||
return start.strftime("%Y-%m")
|
||||
|
||||
|
||||
def _sum_between(events: List[Dict[str, Any]], start: date, end: date,
|
||||
pricing: Dict[str, Any]) -> tuple:
|
||||
lo, hi = start.isoformat(), end.isoformat()
|
||||
evs = [e for e in events if lo <= str(e.get("ts", ""))[:10] < hi]
|
||||
tokens = sum(int(e.get("in", 0) or 0) + int(e.get("out", 0) or 0)
|
||||
+ int(e.get("cache", 0) or 0) for e in evs)
|
||||
cost = sum(cost_usd_events(evs, pricing).values()) if evs else 0.0
|
||||
return tokens, cost
|
||||
|
||||
|
||||
def period_totals(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
offset: int = 0, today: Optional[date] = None) -> tuple:
|
||||
"""(tokens, cost_usd) for the single period ``offset`` periods from now."""
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
return _sum_between(events, start, end, pricing)
|
||||
|
||||
|
||||
def period_window(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
count: int, offset: int = 0, today: Optional[date] = None) -> List[tuple]:
|
||||
"""``count`` consecutive, ZERO-FILLED periods ending at (current + offset),
|
||||
ordered oldest→newest → ``[(label, tokens, cost_usd)]``. ``offset`` (≤ 0)
|
||||
pages the window into the past for the Dashboard's prev/next navigation."""
|
||||
out = []
|
||||
for i in range(count - 1, -1, -1):
|
||||
start, end = period_bounds(gran, offset - i, today)
|
||||
tok, cost = _sum_between(events, start, end, pricing)
|
||||
out.append((_period_label(gran, start), tok, cost))
|
||||
return out
|
||||
|
||||
|
||||
def period_breakdown(events: List[Dict[str, Any]], gran: str, pricing: Dict[str, Any],
|
||||
offset: int = 0, today: Optional[date] = None) -> List[tuple]:
|
||||
"""Break the SELECTED period (``offset`` periods from now) into its sub-parts
|
||||
→ ``[(label, tokens, cost_usd)]``:
|
||||
· week → 7 days Mon→Sun (label ``MM/DD``)
|
||||
· month → weeks W1…Wn (7-day chunks from the 1st)
|
||||
· year → 12 months (label ``01``…``12``)."""
|
||||
from datetime import timedelta
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
out = []
|
||||
if gran == "week":
|
||||
for i in range(7):
|
||||
d = start + timedelta(days=i)
|
||||
tok, cost = _sum_between(events, d, d + timedelta(days=1), pricing)
|
||||
out.append((d.strftime("%m/%d"), tok, cost))
|
||||
elif gran == "year":
|
||||
for m in range(1, 13):
|
||||
ms = date(start.year, m, 1)
|
||||
me = date(start.year + 1, 1, 1) if m == 12 else date(start.year, m + 1, 1)
|
||||
tok, cost = _sum_between(events, ms, me, pricing)
|
||||
out.append((f"{m:02d}", tok, cost))
|
||||
else: # month → weeks W1..Wn
|
||||
ndays = (end - start).days
|
||||
wk, day = 1, 1
|
||||
while day <= ndays:
|
||||
ws = date(start.year, start.month, day)
|
||||
we = date(start.year, start.month, day + 7) if day + 7 <= ndays else end
|
||||
tok, cost = _sum_between(events, ws, we, pricing)
|
||||
out.append((f"W{wk}", tok, cost))
|
||||
wk += 1
|
||||
day += 7
|
||||
return out
|
||||
|
||||
|
||||
def period_range_label(gran: str, offset: int, today: Optional[date] = None) -> str:
|
||||
"""Human label for the selected period (shown in the Dashboard header) —
|
||||
week → MM/DD – MM/DD, month → YYYY/MM, year → YYYY."""
|
||||
from datetime import timedelta
|
||||
start, end = period_bounds(gran, offset, today)
|
||||
if gran == "week":
|
||||
last_day = end - timedelta(days=1)
|
||||
return f"{start.strftime('%m/%d')} – {last_day.strftime('%m/%d')}"
|
||||
if gran == "year":
|
||||
return str(start.year)
|
||||
return start.strftime("%Y/%m")
|
||||
|
||||
|
||||
def set_budget(config, amount: float, currency: Optional[str] = None) -> None:
|
||||
@@ -454,83 +297,11 @@ def budget_status(config) -> Optional[Dict[str, Any]]:
|
||||
}
|
||||
|
||||
|
||||
def format_cost(usd: float, pricing: Dict[str, Any], digits: Optional[int] = None) -> str:
|
||||
"""Format a USD amount in the display currency. ``digits`` caps the number
|
||||
of decimal places (e.g. ``digits=2`` for the Total cost / Budget cards, so
|
||||
USD shows $1.23 not the default up-to-4 $1.2345) — never ADDS decimals to a
|
||||
currency that uses fewer (VND stays whole, JPY one place)."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
cur = p.get("currency", "USD")
|
||||
rate = {"USD": 1.0, "VND": float(p["usd_to_vnd"]), "JPY": float(p["usd_to_jpy"])}.get(cur, 1.0)
|
||||
symbol, cur_digits = _CURRENCY_FMT.get(cur, ("$", 2))
|
||||
if digits is not None:
|
||||
cur_digits = min(cur_digits, digits)
|
||||
value = usd * rate
|
||||
return f"{symbol}{value:,.{cur_digits}f}"
|
||||
|
||||
|
||||
def format_cost_compact(usd: float, pricing: Dict[str, Any]) -> str:
|
||||
"""Compact cost format for the Dashboard chart's y-axis/endpoint labels —
|
||||
always 2 decimals (not format_cost's up-to-4 for USD) and abbreviated with
|
||||
K/M above 1,000/1,000,000, same convention as ``fmt_tokens``. The chart's
|
||||
y-axis label box is narrow; the longer full-precision string used to
|
||||
overflow it, visually clipping/obscuring the leading currency symbol."""
|
||||
p = {**DEFAULT_PRICING, **(pricing or {})}
|
||||
cur = p.get("currency", "USD")
|
||||
rate = {"USD": 1.0, "VND": float(p["usd_to_vnd"]), "JPY": float(p["usd_to_jpy"])}.get(cur, 1.0)
|
||||
symbol, _digits = _CURRENCY_FMT.get(cur, ("$", 2))
|
||||
value = usd * rate
|
||||
sign = "-" if value < 0 else ""
|
||||
value = abs(value)
|
||||
if value >= 1_000_000:
|
||||
body = f"{value / 1_000_000:,.2f}M"
|
||||
elif value >= 1_000:
|
||||
body = f"{value / 1_000:,.2f}K"
|
||||
else:
|
||||
body = f"{value:,.2f}"
|
||||
return f"{sign}{symbol}{body}"
|
||||
|
||||
|
||||
_AI_ANALYSIS_HEADERS = {
|
||||
"vi": ("Nhận xét thói quen", "Cách viết prompt tiết kiệm hơn", "Hành động giảm token"),
|
||||
"en": ("Usage habits", "Writing more efficient prompts", "Actions to cut token usage"),
|
||||
"ja": ("利用傾向", "より効率的なプロンプトの書き方", "トークン削減のためのアクション"),
|
||||
}
|
||||
|
||||
|
||||
def build_ai_analysis_prompt(summary: Dict[str, Any], language: str = "vi") -> str:
|
||||
"""The prompt sent to the model for '✨ AI analyze my usage': aggregated
|
||||
numbers only — never raw prompt contents — asking for concrete habits
|
||||
feedback and token-saving recommendations, in the CURRENTLY SELECTED
|
||||
display language (headers included — not just the model's free-text reply,
|
||||
which would otherwise leave the section titles in Vietnamese regardless of
|
||||
the app's language setting)."""
|
||||
lang_names = {"vi": "Vietnamese", "ja": "Japanese", "en": "English"}
|
||||
h1, h2, h3 = _AI_ANALYSIS_HEADERS.get(language, _AI_ANALYSIS_HEADERS["vi"])
|
||||
top = "\n".join(f"- {label}: {tok:,} tokens"
|
||||
for label, tok in summary.get("top_labels", []))
|
||||
by_source = ", ".join(f"{k}={v:,}" for k, v in summary.get("by_source", []))
|
||||
return (
|
||||
"You are a token-efficiency coach for an AI desktop app (chat tabs + "
|
||||
"scheduled agent tasks). Analyze this usage summary and give the user "
|
||||
"practical advice, replying in "
|
||||
f"{lang_names.get(language, 'Vietnamese')}.\n\n"
|
||||
f"Period stats: {summary.get('turns', 0)} turns, "
|
||||
f"input={summary.get('in', 0):,} tokens, output={summary.get('out', 0):,}, "
|
||||
f"cache={summary.get('cache', 0):,}, "
|
||||
f"avg per prompt={summary.get('avg_per_turn', 0):,}.\n"
|
||||
f"Top consumers:\n{top or '- (none)'}\n"
|
||||
f"By area: {by_source or '(none)'}\n"
|
||||
f"Busiest day: {summary.get('busiest_day')} · busiest hour: {summary.get('busiest_hour')}\n\n"
|
||||
"Reply with EXACTLY these 3 short sections, in markdown, using THESE "
|
||||
f"section headers verbatim (already in {lang_names.get(language, 'Vietnamese')}):\n"
|
||||
f"1. **{h1}** — 2-3 bullet points about the usage pattern.\n"
|
||||
f"2. **{h2}** — 3 concrete prompt-writing tips "
|
||||
"tailored to the numbers above (e.g. long inputs → attach less / summarize "
|
||||
"first; many small turns → batch questions).\n"
|
||||
f"3. **{h3}** — 2-3 app-level actions (compact history, "
|
||||
"smaller model for simple tasks, reuse task outputs instead of re-asking).\n"
|
||||
"Keep the whole reply under 250 words."
|
||||
)
|
||||
|
||||
|
||||
|
||||
Reference in New Issue
Block a user