Repository navigation
Expand file tree
/
Copy pathqoder_proxy.py
More file actions
7850 lines (7274 loc) · 345 KB
/
Copy pathqoder_proxy.py
File metadata and controls
7850 lines (7274 loc) · 345 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""Qoder -> OpenAI-compatible reverse proxy (CN + Intl dual realm).
把 Qoder 国内版 (qoder.com.cn) 与国际版 (qoder.com) 的原生服务封装为标准
OpenAI 兼容接口。上游链路:
客户端 OpenAI 请求
-> build_qoder_body() (官方 baseprompt 模板 + 会话压平)
-> qoder_encode() (Qoder 自定义 Base64 变体)
-> COSY 签名 (RSA 包裹 AES 会话密钥 + MD5 请求签名)
-> POST {gateway}/algo/api/v2/service/pro/sse/agent_chat_generation
-> SSE 信封解包 ({"headers","body","statusCodeValue"} 嵌套帧)
-> 标准 OpenAI SSE / chat.completion 回给客户端
暴露:
GET /v1/models
POST /v1/chat/completions (stream=true / false)
POST /v1/responses (Responses API 双向转换)
GET /health
以及看板全套管理接口 (accounts / usage / tasks / scheduler / settings / logs)
仅使用 Python 标准库(AES/RSA/COSY 签名均为内置纯实现)。
python qoder_proxy.py # bind 127.0.0.1:8790
python qoder_proxy.py --port 9000
python qoder_proxy.py --api-key sk-local # require a bearer token
"""
import argparse
import hashlib
from collections import deque
import re
import json
import os
MAX_PAYLOAD_BYTES = int(os.environ.get("QD_MAX_PAYLOAD_BYTES", 50 * 1024 * 1024)) # 50MB limit
import ipaddress
import socket
import ssl
import sys
import threading
import fnmatch
import time
import urllib.error
import urllib.parse
import urllib.request
import uuid
import qoder_accounts
import qoder_agents
import qoder_anthropic
import qoder_catalog
import qoder_settings
import qoder_sign
from qoder_sign import qoder_encode, SESSIONS
from qoder_accounts import (get_realm_config, gateway_candidates, CLIENT_UA,
local_vm_status)
from pathlib import Path
VERSION = "1.4.0"
CURRENT_REALM = os.environ.get("QD_PROXY_DEFAULT_REALM", "cn")
def is_model_banned(model):
"""P1-4:True = 该模型在本机被全局封禁,不送去上游。
配置源:settings 的 banned_models(fnmatch pattern 列表,默认空 = 不封禁)。
读配置失败一律 fail-open(不误封)。与 wb 的差异:配置在 settings 而非代码
常量——本仓已有面板/设置机制。
"""
name = str(model or "").strip().lower()
if not name:
return False
try:
patterns = qoder_settings.banned_models(ACCOUNTS_DIR)
except Exception:
return False
return any(fnmatch.fnmatchcase(name, p) for p in patterns)
def banned_model_message(model):
"""封禁模型的本机拒绝文案(完全不碰上游)。"""
return ("模型 %s 已被本机网关封禁(settings 的 banned_models 配置)。"
"如需使用,请在设置页或 settings.json 里调整 banned_models。"
% model)
def key_model_message(entry, model):
"""每 Key 模型限制的拒绝文案(与全局封禁同风格)。"""
name = (entry or {}).get("name") or "未命名"
allowed = "、".join((entry or {}).get("models") or []) or "-"
asked = str(model or "").strip() or "(未指定模型)"
return ("API Key「%s」的模型限制不允许 %s。该 Key 目前允许:%s。"
"请在设置页修改该 Key 的模型限制,或改用允许该模型的 Key。"
% (name, asked, allowed))
# 198.18.0.0/15 (RFC 2544 benchmarking) 与 fdfe:dcba:9876::/48 被 Clash/mihomo
# 等本地代理用作 fake-IP DNS 段:开启透明代理的机器上所有公网域名都会解析到
# 这些网段。命中它说明 DNS 已被本机代理接管、真实 IP 不可见,此时跳过解析级
# 校验(名称级校验已完成)。
_FAKEIP_NETS = [
ipaddress.ip_network("198.18.0.0/15"),
ipaddress.ip_network("2001:2::/48"),
ipaddress.ip_network("fdfe:dcba:9876::/48"),
]
def _host_boundary_violation(ip, allow_local):
"""True when the resolved/IP address must be refused."""
if ip.is_loopback:
return not allow_local
if (ip.is_private or ip.is_reserved or ip.is_link_local
or ip.is_multicast or ip.is_unspecified):
return True
return False
def validate_public_http_url(url, allow_local=False):
"""SSRF 防护:仅允许 http/https,且 host 不得指向本机/私有/保留网段。
与 qoder_accounts.validate_public_http_url 同逻辑、独立实现防同错:
本文件的 urlopen 调用点必须经过本文件的边界校验(签名链路上的 URL
在 sess.headers() 之后再校验一次,确保进入 Request 的字符串是洁净的)。
allow_local 仅供显式面向本机网关的开发/验证脚本使用。
"""
parsed = urllib.parse.urlsplit(str(url or ""))
if parsed.scheme not in ("http", "https"):
raise ValueError("only http/https URLs are allowed")
host = (parsed.hostname or "").strip().strip("[]").lower()
if not host:
raise ValueError("URL host is required")
if host == "localhost" or host.endswith(".localhost") or host.endswith(".local"):
if not allow_local:
raise ValueError("requests to localhost are not allowed")
return url
try:
literal = ipaddress.ip_address(host)
except ValueError:
literal = None
if literal is not None:
if _host_boundary_violation(literal, allow_local):
raise ValueError(
"requests to private/reserved address %s are not allowed" % literal)
return url
port = parsed.port or (443 if parsed.scheme == "https" else 80)
try:
infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
except socket.gaierror as exc:
raise ValueError("cannot resolve URL host %r: %s" % (host, exc))
addrs = []
for info in infos:
addr = str(info[4][0]).strip("[]")
try:
addrs.append(ipaddress.ip_address(addr))
except ValueError:
raise ValueError("URL host resolved to a non-IP address: %r" % addr)
if addrs and all(any(ip in net for net in _FAKEIP_NETS) for ip in addrs):
return url # fake-IP DNS:真实 IP 不可见,名称级校验已通过
for ip in addrs:
if _host_boundary_violation(ip, allow_local):
raise ValueError(
"requests to private/reserved address %s are not allowed" % ip)
return url
# 双区自动路由(#26 步1/步2):默认开启。**显式指定永远优先** —— 本开关只作用于
# 「用户没指定区域」时的兜底路径(?realm= / Key 绑定 / X-Realm 任一存在时都不会走到这里)。
AUTO_REALM_ENABLED = os.environ.get("QD_AUTO_REALM", "1").lower() not in (
"0", "false", "no", "off")
def detect_model_realm(model_id):
"""模型 -> 出口区域(**纯数据判断,不查账号可用性**)。
优先级:区域独占硬表 > 清单驱动自动路由(步1)> 全局开关。
清单驱动只在「该模型只在一个区域存在」时换区;双区都有的模型维持当前区
(判不出就不猜 —— 双区同名模型物理上无法靠名字区分,见评估 §2)。
账号可用性不在这里判(那会污染 row_matches_realm 等纯统计路径),
由 fallback_realm_by_accounts() 在选号前单独做(步2)。
"""
owner = exclusive_realm(model_id)
if owner:
return owner
if not model_id:
return CURRENT_REALM
if AUTO_REALM_ENABLED:
realm, reason = qoder_catalog.auto_realm_for_model(model_id, CURRENT_REALM)
if realm:
log("auto realm switch: %s (model=%r, current=%s -> %s)"
% (reason, model_id, CURRENT_REALM, realm), tag="realm")
return realm
return CURRENT_REALM
def auto_realm_reason(model, realm):
"""本次自动换区的**确定性**原因(给面板显示结论,不让它自己推断)。
model_missing 当前区的官方清单里没有这个模型(独占表与清单驱动都归此类)
no_account 模型本身在当前区有(或无法判断),是账号可用性把我们推过去的
用同一份数据源复算(catalog 的 auto_realm_for_model),与 detect_model_realm
同源,不做二次猜测。
"""
try:
hint, _why = qoder_catalog.auto_realm_for_model(model, CURRENT_REALM)
if hint == realm:
return "model_missing"
except Exception:
pass
return "no_account"
def fallback_realm_by_accounts(realm, model):
"""步2:当前区**一个可用账号都没有**时回退另一区(判不出就维持现状)。
只在未显式指定区域时被调用(open_upstream 内)。判据三段:
当前区 count_ready > 0 -> 不动
当前区 == 0 且 另一区 > 0 -> 换区(打日志)
两区都是 0 / 异常 / 无池 -> 不动(交给既有错误路径,不猜)
"""
if not AUTO_REALM_ENABLED or POOL is None:
return realm
if realm != CURRENT_REALM:
# **模型归属优先于账号可用性**:realm 已被模型规则(独占表/清单驱动)推走,
# 说明当前区根本没有这个模型;此时再按账号把请求换回去,只会打到
# 「没有这个模型」的上游。停在目标区、由选号给出「无可用账号」的明确错误
# 才是正确行为。
return realm
try:
if POOL.count_ready(realm, model=model or None) > 0:
return realm
other = "intl" if realm == "cn" else "cn"
if POOL.count_ready(other, model=model or None) > 0:
log("auto realm switch (no ready account): realm %s has no usable "
"account for %r -> %s" % (realm, model or "?", other), tag="realm")
return other
except Exception as exc:
log("auto realm fallback skipped: %s" % exc, level="WARN")
return realm
# 目前双区模型清单不同(源自官方 catalog 快照差集),独占表给区域特供模型。
INTL_EXCLUSIVE_PREFIXES = getattr(qoder_catalog, "INTL_EXCLUSIVE_PREFIXES", ())
CN_EXCLUSIVE_PREFIXES = getattr(qoder_catalog, "CN_EXCLUSIVE_PREFIXES", ())
INTL_EXCLUSIVE = getattr(qoder_catalog, "INTL_EXCLUSIVE", set())
CN_EXCLUSIVE = getattr(qoder_catalog, "CN_EXCLUSIVE", set())
def exclusive_realm(model_id):
""""intl"/"cn" when only that exit serves the model, else "".
先把人类可读别名解析成上游 key 再判独占(glm-5.2 -> gm51model 是国内独占)。
"""
if not model_id:
return ""
m = str(model_id).lower()
resolved = qoder_catalog.resolve_upstream_key(model_id).lower()
for cand in (resolved, m):
if cand in INTL_EXCLUSIVE or cand.startswith(INTL_EXCLUSIVE_PREFIXES):
return "intl"
if cand in CN_EXCLUSIVE or cand.startswith(CN_EXCLUSIVE_PREFIXES):
return "cn"
return ""
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlparse, parse_qs
def install_console_close_handler():
"""Release the port when the console window is closed by the user.
Windows does not kill child processes when a console window closes, so
the proxy (started by the .bat as a child of cmd.exe) would survive and
keep the port bound - the next launch then wrongly reports "another
proxy is already running". Registering an event-driven handler for
CTRL_CLOSE_EVENT is the reliable signal. Harmless without a console.
"""
if os.name != "nt":
return None
try:
import ctypes
from ctypes import wintypes
PHANDLER_ROUTINE = ctypes.WINFUNCTYPE(wintypes.BOOL, wintypes.DWORD)
CTRL_CLOSE_EVENT = 2
CTRL_LOGOFF_EVENT = 5
CTRL_SHUTDOWN_EVENT = 6
def _handler(event):
if event in (CTRL_CLOSE_EVENT, CTRL_LOGOFF_EVENT, CTRL_SHUTDOWN_EVENT):
try:
sys.stdout.flush()
except Exception:
pass
os._exit(0)
return False
handler = PHANDLER_ROUTINE(_handler) # keep the callback referenced
if not ctypes.windll.kernel32.SetConsoleCtrlHandler(handler, True):
return None
return handler
except Exception:
return None
# Launcher names differ per platform; the messages below are read by whoever
# is staring at the terminal, so point at the file that actually exists there.
IS_WINDOWS = os.name == "nt"
START_SCRIPT = "start-qoder-proxy.bat" if IS_WINDOWS \
else "start-qoder-proxy.sh"
FIREWALL_SCRIPT = "allow-firewall.bat" if IS_WINDOWS else "allow-firewall.sh"
def port_owner_hint(port):
"""One-liner(s) the user can paste to see which process holds a port."""
cmds = ["netstat -ano | findstr :%d" % port] if IS_WINDOWS else \
["ss -ltnp | grep :%d" % port,
"lsof -iTCP:%d -sTCP:LISTEN" % port]
return cmds
CHAT_PATH = ("/algo/api/v2/service/pro/sse/agent_chat_generation"
"?FetchKeys=llm_model_result&AgentId=agent_common&Encode=1")
MODELS_PATH = "/algo/api/v2/model/list?Encode=1"
DEFAULT_SYSTEM_PROMPT = "You are a helpful assistant."
# 官方 baseprompt 模板(与桌面端一致的请求体骨架)
BASEPROMPT_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"baseprompt.json")
try:
with open(BASEPROMPT_PATH, encoding="utf-8") as _bp_fh:
BASEPROMPT = json.load(_bp_fh)
except Exception:
BASEPROMPT = {}
NOISE_KEYS = ("extra_fields", "refusal", "reasoning_content")
# DeepSeek 上游 model key(官方目录:dmodel=DeepSeek-V4-Pro / dfmodel=DeepSeek-Flash)。
# 这族模型的多轮一致性与 reasoning_content 强相关,见 is_deepseek_model。
DEEPSEEK_MODEL_KEYS = ("dmodel", "dfmodel")
class BodyTooLarge(Exception):
"""Raised when a request body exceeds the configured cap."""
def __init__(self, length):
super(BodyTooLarge, self).__init__(length)
self.length = length
class BadJSON(Exception):
"""Raised when a request body is present but not a JSON object."""
class UpstreamStatus(Exception):
"""Raised when the upstream SSE envelope carries a non-200 status."""
def __init__(self, status, detail=""):
super(UpstreamStatus, self).__init__("upstream status %s" % status)
self.status = status
self.detail = str(detail or "")
# CORS is only needed by browser-based chat clients that call the OpenAI-style
# API from another origin. Management routes serve the dashboard (same-origin)
# and get no ACAO header - that keeps a stray LAN page from reading them.
CORS_PATH_PREFIXES = ("/v1", "/chat", "/completions", "/models", "/responses")
MANAGEMENT_PATH_PREFIXES = ("/v1/usage", "/usage", "/accounts", "/settings",
"/tasks", "/scheduler", "/panel", "/logs")
def cors_origin_allowed(path):
"""True when the OpenAI-style API path should advertise CORS."""
path = (path or "").split("?")[0]
if path.startswith(MANAGEMENT_PATH_PREFIXES):
return False
return path.startswith(CORS_PATH_PREFIXES)
_lock = threading.Lock()
_login_lock = threading.Lock()
_login_attempts = {} # ip -> list of timestamp
def _prune_login_attempts(now=None, window=60):
"""Drop stale per-IP entries so the dict cannot grow without bound.
Caller must hold _login_lock.
"""
now = now or time.time()
for ip in list(_login_attempts.keys()):
recent = [t for t in _login_attempts[ip] if now - t < window]
if recent:
_login_attempts[ip] = recent
else:
del _login_attempts[ip]
_models_cache = {"intl": {"at": 0.0, "data": None}, "cn": {"at": 0.0, "data": None}}
# Usage accounting: every upstream response carries a usage block, and the
# proxy also records one JSONL line per request.
USAGE_DIR = os.environ.get("QD_PROXY_USAGE_DIR") \
or os.path.join(os.path.dirname(os.path.abspath(__file__)), "usage")
USAGE_LOG = os.path.join(USAGE_DIR, "usage.jsonl")
USAGE_SUMMARY = os.path.join(USAGE_DIR, "usage-summary.json")
DASHBOARD_HTML = os.path.join(os.path.dirname(os.path.abspath(__file__)),
"dashboard.html")
USAGE_FIELDS = ("prompt_tokens", "completion_tokens", "reasoning_tokens",
"cached_tokens", "total_tokens", "credit",
# issue #22:折扣时段对账要用的原价积分(数值,参与聚合)。
# billable 是布尔值,**不做聚合**、只写进 JSONL 行(见 _extract_usage)。
"original_credits")
# Web-panel access control. The panel is gated by its own password (default
# "admin"), independent of the /v1 API key. Sessions live in memory only.
PANEL = qoder_settings.PanelSessions()
API_KEY_FILE_SET = False
def configured_keys():
"""Panel-managed API keys, always read fresh so panel edits apply at once."""
try:
return qoder_settings.api_keys(ACCOUNTS_DIR)
except Exception as exc:
log("could not read api keys: %s" % exc)
return []
def auth_required():
"""Whether /v1 calls must present a key at all."""
if qoder_settings.auth_disabled(ACCOUNTS_DIR):
return False
if any(entry.get("enabled") for entry in configured_keys()):
return True
return bool(API_KEY)
def identify_key(supplied):
"""Return the key entry a caller used, or None when nothing matches.
Once the panel has at least one key, those keys are the only accepted
credentials - otherwise a launcher key left in a .bat file would silently
keep working after the panel was locked down.
"""
extra = () if configured_keys() else (API_KEY,)
return qoder_settings.match_api_key(ACCOUNTS_DIR, supplied, extra_keys=extra)
def _empty_stats():
return {"requests": 0, "errors": 0, "prompt_tokens": 0,
"completion_tokens": 0, "reasoning_tokens": 0, "cached_tokens": 0,
"total_tokens": 0, "credit": 0.0, "original_credits": 0.0,
"started": time.time(),
"by_model": {},
"ttft_ms_sum": 0, "ttft_samples": 0,
"gen_ms_sum": 0, "gen_samples": 0,
"wall_ms_sum": 0, "wall_samples": 0}
_usage = _empty_stats()
def _extract_usage(usage):
"""Normalize the upstream usage block into the fields we track."""
if not usage:
return {}
details = usage.get("completion_tokens_details") or {}
prompt_details = usage.get("prompt_tokens_details") or {}
return {
"prompt_tokens": usage.get("prompt_tokens") or 0,
"completion_tokens": usage.get("completion_tokens") or 0,
"reasoning_tokens": details.get("reasoning_tokens") or 0,
"cached_tokens": usage.get("prompt_cache_hit_tokens")
or details.get("cached_tokens") or prompt_details.get("cached_tokens") or 0,
"total_tokens": usage.get("total_tokens") or 0,
# issue #22:上游给的是**复数 credits** —— 旧实现只读单数 credit,
# 恒为 None -> 所有 usage 记录成 0(历史数据无法回填,原始值未落盘)。
# 保留单数兜底以兼容其它区域/版本的字段差异。
"credit": usage.get("credits") or usage.get("credit") or 0,
"original_credits": usage.get("original_credits") or 0,
# billable 只记录不聚合(布尔求和无意义):进 JSONL 行,不进
# USAGE_FIELDS 的累加与 by_model。
"billable": bool(usage.get("billable")),
}
def realm_scope(value):
"""面板 realm 取值的归一:显式 "all"(大小写不敏感)= 全部区域 → None(无过滤)。
只有字面 "all" 被当作通配:None / "" 本来就表示"无过滤"(各路由会先回落到
CURRENT_REALM 再传进来),"cn" / "intl" / 其它未知值**一律原样保留既有语义**
(按字面比较;未知值自然过滤出空集 —— 不在这里替面板猜意图)。
"""
v = str(value if value is not None else "").strip()
return None if v.lower() in ("", "all") else v
def row_matches_realm(row, realm):
realm = realm_scope(realm)
if not realm:
return True
r = row.get("realm")
if r:
return r == realm
acct_uid = row.get("account")
if acct_uid and POOL:
acc = POOL.get(acct_uid)
if acc:
return acc.realm == realm
model = row.get("model")
if model:
return detect_model_realm(model) == realm
return realm == "cn"
def record_usage(model, usage, stream=None, elapsed_ms=None, ttft_ms=None,
gen_ms=None, fp=None, account=None, effort=None,
effort_requested=None, key_id=None, realm_reason=None):
"""Accumulate stats, append a JSONL row, and persist the summary.
P1-3(推理档位审计 + key 归属):后三个参数都是**可选**的,旧调用路径不传
就一个键都不写 —— 行格式与改动前逐字节一致。
effort 实际下发给上游的档位(None = 本次没有下发该参数)
effort_requested 客户端请求的原始档位(归一化之前)
key_id API key 的**标识**(panel id / legacy / launcher),不涉明文
请求档位与实际档位同时落盘,才能看出归一化有没有把它们改掉/丢掉。
"""
fields = _extract_usage(usage)
if not fields:
return None
row = {
"at": time.time(),
"iso": time.strftime("%Y-%m-%dT%H:%M:%S"),
"model": model,
"stream": bool(stream),
"elapsed_ms": elapsed_ms,
"ttft_ms": ttft_ms,
"gen_ms": gen_ms,
}
row.update(fields)
if fp:
row.update(fp)
if account:
row["account"] = account
# P1-3:推理档位审计(请求档位 vs 实际下发档位)与 API key 归属。
# 三者都是可选参数:不传则一个键都不写,旧行格式不变。
if effort_requested:
row["reasoning_effort_requested"] = str(effort_requested)
if effort:
row["reasoning_effort"] = str(effort)
elif effort_requested:
# 请求了档位但归一化后不下发(模型无档位表 / 被丢弃):显式记空串,
# 便于区分「客户端没请求」与「请求了但没生效」两种情况。
row["reasoning_effort"] = ""
if key_id:
# 只记 key 的**标识**(panel 分配的 id / legacy / launcher),绝无明文 key。
row["key_id"] = str(key_id)
if realm_reason is not None:
# #26:本次是否**自动换区**及原因(持久化,面板不依赖响应头即可读到)。
# "model_missing" 当前区官方清单里没有这个模型(独占表 / 清单驱动)
# "no_account" 模型本区有(或无法判断),是账号可用性把它推走的
# ""(空串) 本次**没有换区**(正常路径)
# 键不存在 该行产生于本字段上线前(旧行)—— 与「没换区」明确区分
row["realm_reason"] = str(realm_reason)
acc = POOL.get(account) if (account and POOL) else None
row["realm"] = acc.realm if acc else CURRENT_REALM
if gen_ms and gen_ms > 0:
row["tokens_per_sec"] = round(
fields["completion_tokens"] / (gen_ms / 1000.0), 2)
if fields["prompt_tokens"] > 0:
row["cache_hit_pct"] = round(
fields["cached_tokens"] * 100.0 / fields["prompt_tokens"], 1)
with _lock:
_usage["requests"] += 1
for k in USAGE_FIELDS:
if k in fields:
_usage[k] += fields[k]
if ttft_ms is not None:
_usage["ttft_ms_sum"] += ttft_ms
_usage["ttft_samples"] += 1
if gen_ms is not None:
_usage["gen_ms_sum"] += gen_ms
_usage["gen_samples"] += 1
if elapsed_ms is not None:
_usage["wall_ms_sum"] += elapsed_ms
_usage["wall_samples"] += 1
per = _usage["by_model"].setdefault(
model, {"requests": 0, **{k: 0 for k in USAGE_FIELDS}})
per["requests"] += 1
for k in USAGE_FIELDS:
if k in fields:
per[k] += fields[k]
summary = json.loads(json.dumps(_usage))
_persist_usage(row, summary, "usage persist failed")
try:
t_tokens = fields.get("total_tokens", 0)
dur = " %dms" % elapsed_ms if elapsed_ms is not None else ""
acc_tag = " acct=%s" % account[:8] if account else ""
speed_tag = " %st/s" % row.get("tokens_per_sec", 0) \
if row.get("tokens_per_sec") else ""
log("chat done: model=%s%s%s tokens=%d (in=%d out=%d)%s"
% (model, acc_tag, dur, t_tokens, fields.get("prompt_tokens", 0),
fields.get("completion_tokens", 0), speed_tag), tag="chat")
except Exception:
pass
return row
def _persist_usage(row, summary, fail_label):
"""Append one JSONL row and atomically rewrite the summary.
The temp file carries a unique suffix: two threads writing the same
"<summary>.tmp" race, and the loser's os.replace() fails with ENOENT
because the winner already renamed the file away.
"""
try:
usage_dir = Path(USAGE_DIR).resolve()
usage_dir.mkdir(parents=True, exist_ok=True)
log_path = usage_dir / os.path.basename(USAGE_LOG)
summary_path = usage_dir / os.path.basename(USAGE_SUMMARY)
if not (log_path.is_relative_to(usage_dir)
and summary_path.is_relative_to(usage_dir)):
raise ValueError("usage path escapes base directory")
with open(log_path, "a", encoding="utf-8") as fh:
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
tmp_summary = usage_dir / (
os.path.basename(USAGE_SUMMARY)
+ ".%d.%d.tmp" % (os.getpid(), threading.get_ident()))
if not tmp_summary.is_relative_to(usage_dir):
raise ValueError("usage path escapes base directory")
try:
tmp_summary.write_text(json.dumps(summary, ensure_ascii=False,
indent=2), encoding="utf-8")
os.replace(tmp_summary, summary_path)
except Exception:
try:
os.unlink(tmp_summary)
except Exception:
pass
raise
except Exception as exc:
log("%s: %s" % (fail_label, exc))
def record_error(model, status, message, elapsed_ms=None):
"""Count a failed request and append it to the log so errors are visible."""
row = {
"at": time.time(),
"iso": time.strftime("%Y-%m-%dT%H:%M:%S"),
"model": model,
"error": True,
"status": status,
# 保留更完整的上游错误(provider_error 的内层 details 常在 200+ 字节)
"message": str(message)[:400],
"elapsed_ms": elapsed_ms,
}
with _lock:
_usage["errors"] += 1
if elapsed_ms is not None:
_usage["wall_ms_sum"] += elapsed_ms
_usage["wall_samples"] += 1
summary = json.loads(json.dumps(_usage))
_persist_usage(row, summary, "error persist failed")
dur = " %dms" % elapsed_ms if elapsed_ms is not None else ""
log("request error: model=%s%s status=%s msg=%s"
% (model, dur, status, str(message)[:360]),
level="ERROR", tag="chat")
return row
def _pct(values, q):
"""Nearest-rank percentile (no interpolation) - good enough for latency."""
if not values:
return None
ordered = sorted(values)
idx = int(round((q / 100.0) * (len(ordered) - 1)))
return ordered[max(0, min(len(ordered) - 1, idx))]
# ---------------------------------------------------------------------------
# 时间窗口:?range=day|week|month|all|custom&since=&until=
# ---------------------------------------------------------------------------
# 语义(与面板选择器、wb 参考实现一致):day/week/month 是**锚到本地零点**的日历
# 窗口(week 从周一起算、month 从 1 号起算),custom 取面板传来的两个 epoch(任一端
# 可缺省,反了自动交换),all / 空 / 未知值 = 不设窗口(历史全量)。返回 (since, until),
# None 表示该侧无界;两侧都 None 时下游**完全不做过滤**(_row_in_window 直接返回 True,
# 连 at 都不读),因此不传 range 的输出与改动前逐字节一致。
def _epoch_or_none(value):
"""面板传来的 epoch 秒;负值/不可解析一律丢弃(不做钳制,宁可不设界)。"""
if value in (None, "", False):
return None
try:
seconds = float(value)
except (TypeError, ValueError):
return None
return seconds if seconds >= 0 else None
def _local_midnight(days_back=0):
lt = time.localtime(time.time() - days_back * 86400)
return time.mktime((lt.tm_year, lt.tm_mon, lt.tm_mday, 0, 0, 0, 0, 0, -1))
def range_window(value, since=None, until=None):
v = str(value or "").strip().lower()
if v in ("today", "day", "1d"):
return _local_midnight(), None
if v in ("week", "w"):
return _local_midnight(days_back=time.localtime().tm_wday), None
if v in ("month", "m"):
lt = time.localtime()
return time.mktime((lt.tm_year, lt.tm_mon, 1, 0, 0, 0, 0, 0, -1)), None
if v == "custom":
lo, hi = _epoch_or_none(since), _epoch_or_none(until)
if lo is not None and hi is not None and hi < lo:
lo, hi = hi, lo
return lo, hi
return None, None
def range_query(query):
"""从 parse_qs 的查询串里取 (range, since, until);缺省一律 None。"""
def first(name):
values = query.get(name) or [None]
return values[0] if values else None
return first("range"), first("since"), first("until")
def _row_in_window(row, since, until):
"""窗口过滤:两侧都 None 时不做任何读取(老路径零开销、零差异)。"""
if since is None and until is None:
return True
at = row.get("at")
if not isinstance(at, (int, float)):
at = 0
if since is not None and at < since:
return False
if until is not None and at > until:
return False
return True
_perf_cache = {}
_perf_lock = threading.Lock()
def perf_stats(sample=5000, realm=None, ttl=10, range=None, since=None, until=None):
"""Cached wrapper: parsing thousands of rows is CPU-heavy, and the
dashboard polls this endpoint every few seconds."""
r = realm or CURRENT_REALM
lo, hi = range_window(range, since, until)
try:
# 缓存键带上**解析后的边界**(不是 range 标签):week 与 month 相互重叠,
# 一个标签无法描述两个窗口,否则会拿一个窗口的数据冒充另一个。
key = (int(sample), r, lo, hi)
except Exception:
key = (5000, r, lo, hi)
now = time.time()
with _perf_lock:
hit = _perf_cache.get(key)
if hit is not None and (now - hit[0]) < ttl:
return hit[1]
data = _perf_stats_uncached(sample, realm, since=lo, until=hi)
with _perf_lock:
_perf_cache[key] = (time.time(), data)
return data
def _perf_stats_uncached(sample=5000, realm=None, since=None, until=None):
"""Latency percentiles + derived rates, computed from the JSONL log.
since/until 为窗口边界(None=该侧无界);两者都 None 时行为与改动前一致。
"""
ttfts, gens, walls, rates, hits, tok_rates = [], [], [], [], [], []
total = ok = err = 0
m_buckets = {}
# 只读日志末尾 sample 行:readlines() 会把整个日志读成字符串列表。
rows = [raw.decode("utf-8", "replace") for raw in _tail_lines(USAGE_LOG, sample)]
for line in rows:
line = line.strip()
if not line:
continue
try:
r = json.loads(line)
except Exception:
continue
if realm and not row_matches_realm(r, realm):
continue
if not _row_in_window(r, since, until):
continue
total += 1
if r.get("error"):
err += 1
if r.get("elapsed_ms"):
walls.append(r["elapsed_ms"])
continue
ok += 1
if r.get("ttft_ms") is not None:
ttfts.append(r["ttft_ms"])
if r.get("gen_ms") is not None:
gens.append(r["gen_ms"])
if r.get("elapsed_ms") is not None:
walls.append(r["elapsed_ms"])
if r.get("tokens_per_sec"):
tok_rates.append(r["tokens_per_sec"])
if r.get("cache_hit_pct") is not None:
hits.append(r["cache_hit_pct"])
m_id = r.get("model") or "unknown"
mb = m_buckets.setdefault(m_id, {"total": 0, "ok": 0, "err": 0,
"ttfts": [], "gens": [], "walls": [],
"tok_rates": [], "hits": []})
mb["total"] += 1
if r.get("error"):
mb["err"] += 1
else:
mb["ok"] += 1
if r.get("ttft_ms") is not None:
mb["ttfts"].append(r["ttft_ms"])
if r.get("gen_ms") is not None:
mb["gens"].append(r["gen_ms"])
if r.get("elapsed_ms") is not None:
mb["walls"].append(r["elapsed_ms"])
if r.get("tokens_per_sec"):
mb["tok_rates"].append(r["tokens_per_sec"])
if r.get("cache_hit_pct") is not None:
mb["hits"].append(r["cache_hit_pct"])
def block(vals):
if not vals:
return None
return {
"avg": round(sum(vals) / len(vals), 1),
"p50": _pct(vals, 50),
"p90": _pct(vals, 90),
"p99": _pct(vals, 99),
"max": max(vals),
"samples": len(vals),
}
return {
"sampled": total,
"success": ok,
"errors": err,
"inner_errors": inner_error_snapshot(),
"success_rate_pct": round(ok * 100.0 / total, 1) if total else None,
"ttft_ms": block(ttfts),
"generation_ms": block(gens),
"wall_ms": block(walls),
"tokens_per_sec": block(tok_rates),
"cache_hit_pct": block(hits),
"by_model": {
mid: {
"requests": mb["total"],
"errors": mb["err"],
"success_rate_pct": round(mb["ok"] * 100.0 / mb["total"], 1)
if mb["total"] else None,
"ttft_ms": block(mb["ttfts"]),
"generation_ms": block(mb["gens"]),
"wall_ms": block(mb["walls"]),
"tokens_per_sec": block(mb["tok_rates"]),
"cache_hit_pct": block(mb["hits"]),
} for mid, mb in m_buckets.items()
},
}
_snap_cache = {}
_snap_lock = threading.Lock()
def usage_snapshot(realm=None, ttl=10):
"""Cached wrapper: the dashboard polls this every few seconds."""
r = realm or CURRENT_REALM
now = time.time()
with _snap_lock:
hit = _snap_cache.get(r)
if hit is not None and (now - hit[0]) < ttl:
return hit[1]
data = _usage_snapshot_uncached(r)
with _snap_lock:
_snap_cache[r] = (time.time(), data)
return data
def _usage_snapshot_uncached(realm=None):
r = realm or CURRENT_REALM
rep = POOL.representative(realm=r) if POOL else current_account()
snap = _empty_stats()
snap["started"] = _usage.get("started", time.time())
try:
with open(USAGE_LOG, encoding="utf-8") as fh:
for line in fh:
line = line.strip()
if not line:
continue
try:
row = json.loads(line)
except Exception:
continue
if r and not row_matches_realm(row, r):
continue
if row.get("error"):
snap["errors"] += 1
else:
snap["requests"] += 1
for k in USAGE_FIELDS:
if k in row:
snap[k] += (row[k] or 0)
m = row.get("model") or "unknown"
per = snap["by_model"].setdefault(
m, {"requests": 0, "accounts": {},
**{k: 0 for k in USAGE_FIELDS}})
per["requests"] += 1
for k in USAGE_FIELDS:
if k in row:
per[k] += (row[k] or 0)
acct_id = row.get("account")
if acct_id:
per.setdefault("accounts", {})
per["accounts"][acct_id] = per["accounts"].get(acct_id, 0) + 1
except FileNotFoundError:
pass
except Exception as exc:
log("usage snapshot read failed: %s" % exc)
snap["since"] = time.strftime("%Y-%m-%d %H:%M:%S",
time.localtime(snap.get("started", time.time())))
snap["log_file"] = USAGE_LOG
snap["realm"] = r
snap["accounts_map"] = {a.uid: {"nickname": a.nickname, "realm": a.realm}
for a in POOL.accounts} if POOL else {}
snap["account"] = {
"uid": (rep.uid if rep else ""),
"domain": (rep.domain if rep else ""),
"issuer": ("qoder" if rep else ""),
"credential_file": (os.path.basename(rep.path) if rep and rep.path else ""),
"expires_at": (rep.expires_at if rep else 0),
"accounts": (len(POOL.accounts) if POOL else 0),
"accounts_ready": (POOL.count_ready() if POOL else 0),
}
return snap
def _tail_lines(path, max_lines, chunk=256 * 1024):
"""Return up to the last `max_lines` non-empty lines, oldest first.
The usage log passes 20MB within a day. Scanning it end to end on every
dashboard poll was the dominant cost behind slow /usage/* responses.
"""
lines = []
try:
with open(path, "rb") as fh:
fh.seek(0, os.SEEK_END)
pos = fh.tell()
buf = b""
while pos > 0 and len(lines) < max_lines:
step = min(chunk, pos)
pos -= step
fh.seek(pos)
buf = fh.read(step) + buf
parts = buf.split(b"\n")
buf = parts[0]
for raw in reversed(parts[1:]):
if not raw.strip():
continue
lines.append(raw)
if len(lines) >= max_lines:
break
if len(lines) < max_lines and buf.strip():
lines.append(buf)
except FileNotFoundError:
return []
except Exception as exc:
log("tail read failed: %s" % exc)
return []
lines.reverse()
return lines
def count_usage_rows(realm=None):
"""Cheap row count - substring match instead of a full JSON parse."""
realm = realm_scope(realm) # realm=all 走无 needles 分支,与逐行过滤一致
needles = ()
if realm:
needles = ('"realm": "%s"' % realm, '"realm":"%s"' % realm)
n = 0
try:
with open(USAGE_LOG, encoding="utf-8") as fh:
for line in fh:
if not line.strip():
continue
if not needles:
n += 1
continue
if any(x in line for x in needles):
n += 1
continue
if '"realm"' in line:
continue
try:
if row_matches_realm(json.loads(line), realm):