"""pipeline_llm 模块入口 — 模型治理（供应商/账号/模型/组织策略/限额限流/记账）。

宿主集成：应用入口调用 load_pipeline_llm()（try/except ImportError 兜底）。
所有表共享宿主应用的库（get_module_dbname('pipeline_llm')），表名统一 llm_ 前缀。

模块只提供治理引擎与配置管理；实际 LLM 转发仍由宿主已有的
pipeline_service.llm_bridge / llm_proxy 承担——本模块在两个调用链上挂钩子：
  - llm_bridge._get_model_config：治理启用时前置解析（主备容错+端点选择）
  - llm_proxy.proxy_chat_completion：调用前后接门禁与结算（运行环境路径）
"""

import html
import json
import logging
import os

from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv
from appPublic.uniqueID import getID

from .gateway import (
    govern_resolve, govern_settle, governance_enabled,
    recharge_account, recharge_org,
    encrypt_api_key, decrypt_api_key, load_gateway, GovernError,
)
from .inference import chat_inference, list_models, load_inference
from .selection import load_selection
from .product_interface import load_product_category_product

logger = logging.getLogger("pipeline_llm")

MODULE_NAME = 'pipeline_llm'


def _dbname():
    """动态取宿主库名（禁硬编码）。未注册 get_module_dbname 时回落 pipeline。"""
    env = ServerEnv()
    fn = getattr(env, 'get_module_dbname', None)
    if callable(fn):
        try:
            return fn(MODULE_NAME) or 'pipeline'
        except Exception:
            pass
    return 'pipeline'


def _get_sor():
    return DBPools(), _dbname()


# ────────────────────── 端点目录输入归一化 ──────────────────────

def normalize_endpoints(text):
    """端点目录输入归一化：支持「一行一个 URL」或 JSON 数组两种写法。

    行格式：URL [region=domestic|international] [timeout=秒]，例如：
        https://dashscope.aliyuncs.com
        https://api.example.com region=international timeout=30
    缺省 region=domestic、timeout=60。存储格式恒为 JSON 数组
    [{base_url, region, timeout}]（治理链按此读取）。
    非法输入抛 ValueError（中文提示，直接展示给用户）。

    2026-09-12 端点可互换约定：base_url 一律归一为主机根（scheme+域名）——
    输入带路径段（旧数据/旧习惯的 /api/v1、/compatible-mode/v1 等）时确定性
    剥掉；接口路径归 profile.path 承载（完整相对路径）。同 (主机根, region)
    去重合并，timeout 取最大值（单端点须覆盖该主机全部接口形态的超时需求）。
    """
    from urllib.parse import urlparse
    s = (text or '').strip()
    if not s:
        return ''
    if s.startswith('['):
        eps = json.loads(s)
        if not isinstance(eps, list):
            raise ValueError('endpoints 必须是 JSON 数组')
    else:
        eps = []
        for ln_no, ln in enumerate([l.strip() for l in s.splitlines()], 1):
            if not ln:
                continue
            parts = ln.split()
            url = parts[0].rstrip('/')
            if not (url.startswith('http://') or url.startswith('https://')):
                raise ValueError('第 %d 行不是合法 URL（须以 http(s):// 开头）：%s' % (ln_no, parts[0]))
            ep = {'base_url': url, 'region': 'domestic', 'timeout': 60}
            for kv in parts[1:]:
                if '=' not in kv:
                    raise ValueError('第 %d 行选项 %s 应为 key=value 形式（region= / timeout=）' % (ln_no, kv))
                k, v = kv.split('=', 1)
                if k == 'region':
                    if v not in ('domestic', 'international'):
                        raise ValueError('第 %d 行 region 只能是 domestic 或 international' % ln_no)
                    ep['region'] = v
                elif k == 'timeout':
                    ep['timeout'] = int(v)
                else:
                    raise ValueError('第 %d 行未知选项 %s（支持 region= / timeout=）' % (ln_no, k))
            eps.append(ep)
    merged = []
    for i, ep in enumerate(eps):
        if not isinstance(ep, dict) or not (ep.get('base_url') or '').strip():
            raise ValueError('端点 #%d 缺少 base_url' % (i + 1))
        pu = urlparse((ep.get('base_url') or '').rstrip('/'))
        if not pu.netloc:
            raise ValueError('端点 #%d base_url 非法（无主机名）：%s' % (i + 1, ep.get('base_url')))
        root = pu.scheme + '://' + pu.netloc
        region = ep.get('region') or 'domestic'
        try:
            to = int(ep.get('timeout') or 60)
        except (TypeError, ValueError):
            to = 60
        hit = None
        for m in merged:
            if m['base_url'] == root and m['region'] == region:
                hit = m
                break
        if hit is None:
            m = {'base_url': root, 'region': region, 'timeout': to}
            if ep.get('protocol'):
                m['protocol'] = ep['protocol']
            merged.append(m)
        else:
            hit['timeout'] = max(hit['timeout'], to)
    return json.dumps(merged, ensure_ascii=False)


def _norm_endpoint_ids(raw):
    """endpoint_ids 归一化：JSON 数组或逗号分隔数字均可，返回 JSON 数组串。"""
    if isinstance(raw, (list, dict)):
        raw = json.dumps(raw, ensure_ascii=False)
    raw = (raw or '').strip()
    if not raw:
        return ''
    if raw.startswith('['):
        idxs = json.loads(raw)
    else:
        idxs = [x.strip() for x in raw.split(',') if x.strip()]
    if not isinstance(idxs, list):
        raise ValueError('endpoint_ids 必须是 JSON 数组')
    for i in idxs:
        if not (isinstance(i, int) or (isinstance(i, str) and i.lstrip('-').isdigit())):
            raise ValueError('endpoint_ids 元素必须是端点下标（数字）')
    return json.dumps([int(i) for i in idxs], ensure_ascii=False)


MASK = '******'  # 列表脱敏占位；编辑表单原样提交视为「不修改」


# ────────────────────── CRUD（供生成的 CRUD 页面调用） ──────────────────────

def _clean(params_kw):
    data = dict(params_kw or {})
    for k in ('page', 'rows', 'data_filter', 'sortby'):
        data.pop(k, None)
    return data


async def create_llm_vendor(params_kw):
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        data['endpoints'] = normalize_endpoints(data.get('endpoints', ''))
        data['id'] = getID()
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            await sor.C('llm_vendor', data)
        result['success'] = True
        result['message'] = '创建成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def update_llm_vendor(params_kw):
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        if 'endpoints' in data:
            data['endpoints'] = normalize_endpoints(data.get('endpoints', ''))
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            await sor.U('llm_vendor', data)
        result['success'] = True
        result['message'] = '更新成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def delete_llm_vendor(params_kw):
    """供应商删除前置检查：有账号/模型引用时禁止（防孤儿）。"""
    result = {'success': False, 'message': ''}
    try:
        vid = (params_kw or {}).get('id', '')
        if not vid:
            raise ValueError('缺少 id')
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT (SELECT COUNT(*) FROM llm_account WHERE vendor_id=${v}$) + "
                "(SELECT COUNT(*) FROM llm_model WHERE vendor_id=${v}$) AS c", {"v": vid})
            await sor.sqlExe("COMMIT", {})
            cnt = int(getattr(recs[0], 'c', 0)) if recs else 0
            if cnt > 0:
                raise ValueError('该供应商下还有 %d 个账号/模型，请先移除后再删除' % cnt)
            await sor.sqlExe("DELETE FROM llm_vendor WHERE id=${v}$", {"v": vid})
            await sor.sqlExe("COMMIT", {})
        result['success'] = True
        result['message'] = '删除成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def create_llm_account(params_kw):
    """账号创建：api_key 自动 AES 加密；校验端点下标合法。"""
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        vid = data.get('vendor_id', '')
        if not vid:
            raise ValueError('必须选择供应商')
        # api_key 加密存储（掩码/空 = 不设置）
        if data.get('api_key') and data['api_key'] != MASK:
            data['api_key'] = encrypt_api_key(data['api_key'])
        else:
            data.pop('api_key', None)
        # 端点下标校验（JSON 或逗号分隔均可）
        data['endpoint_ids'] = _norm_endpoint_ids(data.get('endpoint_ids', ''))
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe("SELECT endpoints FROM llm_vendor WHERE id=${v}$", {"v": vid})
            await sor.sqlExe("COMMIT", {})
            if not recs:
                raise ValueError('供应商不存在')
            eps = []
            try:
                eps = json.loads(getattr(recs[0], 'endpoints', '') or '') or []
            except Exception:
                eps = []
            for i in json.loads(data['endpoint_ids'] or '[]'):
                if int(i) < 0 or int(i) >= len(eps):
                    raise ValueError('端点下标 %s 超出供应商端点目录（共 %d 个端点）' % (i, len(eps)))
        data['id'] = getID()
        async with db.sqlorContext(dbname) as sor:
            await sor.C('llm_account', data)
        result['success'] = True
        result['message'] = '创建成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def update_llm_account(params_kw):
    """账号更新：api_key 非空才重新加密（空=不改）。

    vendor_id 归属不可改：账号在哪个供应商下创建就固定在哪个供应商下，
    防止子表（按供应商过滤）入口下篡改表单把账号挪到别的供应商。
    """
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        aid = data.get('id', '')
        if not aid:
            raise ValueError('缺少 id')
        data.pop('vendor_id', None)
        if data.get('api_key') and data['api_key'] != MASK:
            data['api_key'] = encrypt_api_key(data['api_key'])
        else:
            data.pop('api_key', None)
        if 'endpoint_ids' in data:
            data['endpoint_ids'] = _norm_endpoint_ids(data.get('endpoint_ids', ''))
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            await sor.U('llm_account', data)
        result['success'] = True
        result['message'] = '更新成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def delete_llm_account(params_kw):
    """账号删除：被模型引用时禁止；带 vendor_id 时校验归属（子表范围隔离）。"""
    result = {'success': False, 'message': ''}
    try:
        pk = params_kw or {}
        aid = pk.get('id', '')
        if not aid:
            raise ValueError('缺少 id')
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT vendor_id FROM llm_account WHERE id=${a}$", {"a": aid})
            await sor.sqlExe("COMMIT", {})
            if not recs:
                raise ValueError('账号不存在')
            # 子表入口会带 vendor_id：校验账号确实属于该供应商，防跨供应商误删/越权删
            expect_vid = pk.get('vendor_id', '')
            if expect_vid and getattr(recs[0], 'vendor_id', '') != expect_vid:
                raise ValueError('账号不属于该供应商，禁止删除')
            cnt_recs = await sor.sqlExe(
                "SELECT COUNT(*) AS c FROM llm_model WHERE account_id=${a}$", {"a": aid})
            await sor.sqlExe("COMMIT", {})
            cnt = int(getattr(cnt_recs[0], 'c', 0)) if cnt_recs else 0
            if cnt > 0:
                raise ValueError('该账号被 %d 个模型设为默认账号，请先解除后再删除' % cnt)
            await sor.sqlExe("DELETE FROM llm_account WHERE id=${a}$", {"a": aid})
            await sor.sqlExe("COMMIT", {})
        result['success'] = True
        result['message'] = '删除成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def create_llm_model(params_kw):
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        name = (data.get('name', '') or '').strip()
        if not name:
            raise ValueError('模型注册名不能为空')
        if not data.get('vendor_id'):
            raise ValueError('必须选择供应商')
        dp = data.get('default_params', '') or ''
        if isinstance(dp, (list, dict)):
            dp = json.dumps(dp, ensure_ascii=False)
        if dp.strip():
            json.loads(dp)  # 校验合法
        data['default_params'] = dp
        # 后续步骤模板链：JSON 数组校验
        qp = data.get('query_profile_ids', '') or ''
        if isinstance(qp, (list, dict)):
            qp = json.dumps(qp, ensure_ascii=False)
        if qp.strip():
            ql = json.loads(qp)
            if not isinstance(ql, list):
                raise ValueError('query_profile_ids 必须是 JSON 数组')
        data['query_profile_ids'] = qp
        data['id'] = getID()
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe("SELECT id FROM llm_model WHERE name=${n}$", {"n": name})
            await sor.sqlExe("COMMIT", {})
            if recs:
                raise ValueError('模型注册名「%s」已存在（平台内唯一）' % name)
            await sor.C('llm_model', data)
        result['success'] = True
        result['message'] = '创建成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def update_llm_model(params_kw):
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        if 'default_params' in data:
            dp = data.get('default_params', '') or ''
            if isinstance(dp, (list, dict)):
                dp = json.dumps(dp, ensure_ascii=False)
            if dp.strip():
                json.loads(dp)
            data['default_params'] = dp
        if 'query_profile_ids' in data:
            qp = data.get('query_profile_ids', '') or ''
            if isinstance(qp, (list, dict)):
                qp = json.dumps(qp, ensure_ascii=False)
            if qp.strip():
                ql = json.loads(qp)
                if not isinstance(ql, list):
                    raise ValueError('query_profile_ids 必须是 JSON 数组')
            data['query_profile_ids'] = qp
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            await sor.U('llm_model', data)
        result['success'] = True
        result['message'] = '更新成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def delete_llm_model(params_kw):
    """模型删除前置检查：被策略引用时禁止。"""
    result = {'success': False, 'message': ''}
    try:
        mid = (params_kw or {}).get('id', '')
        if not mid:
            raise ValueError('缺少 id')
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT COUNT(*) AS c FROM llm_org_policy WHERE primary_model_id=${m}$", {"m": mid})
            await sor.sqlExe("COMMIT", {})
            cnt = int(getattr(recs[0], 'c', 0)) if recs else 0
            if cnt > 0:
                raise ValueError('该模型被 %d 个组织策略设为主模型，请先调整策略' % cnt)
            await sor.sqlExe("DELETE FROM llm_model WHERE id=${m}$", {"m": mid})
            await sor.sqlExe("COMMIT", {})
        result['success'] = True
        result['message'] = '删除成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def update_llm_org_policy(params_kw):
    """组织策略保存：校验模型存在、备链去重。"""
    result = {'success': False, 'message': ''}
    try:
        data = _clean(params_kw)
        backups = data.get('backup_model_ids', '') or ''
        if isinstance(backups, (list, dict)):
            backups = json.dumps(backups, ensure_ascii=False)
        bl = []
        if backups.strip():
            bl = json.loads(backups)
            if not isinstance(bl, list):
                raise ValueError('backup_model_ids 必须是 JSON 数组')
            seen = []
            for b in bl:
                if b and b not in seen:
                    seen.append(b)
            bl = seen
        data['backup_model_ids'] = json.dumps(bl, ensure_ascii=False) if bl else ''
        primary = data.get('primary_model_id', '') or ''
        if primary and primary in bl:
            raise ValueError('主模型不能同时出现在备模型链中')
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            if primary:
                recs = await sor.sqlExe("SELECT id FROM llm_model WHERE id=${m}$", {"m": primary})
                await sor.sqlExe("COMMIT", {})
                if not recs:
                    raise ValueError('主模型不存在或已停用')
            org_id = data.get('org_id', '')
            if org_id:
                recs = await sor.sqlExe("SELECT id FROM llm_org_policy WHERE org_id=${o}$", {"o": org_id})
                await sor.sqlExe("COMMIT", {})
                if recs:
                    data['id'] = getattr(recs[0], 'id', '')
                    await sor.U('llm_org_policy', data)
                else:
                    data['id'] = getID()
                    await sor.C('llm_org_policy', data)
            else:
                await sor.U('llm_org_policy', data)
        result['success'] = True
        result['message'] = '保存成功'
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def llm_usage_query(params_kw):
    """用量查询（列表页工具栏）：按机构/模型/账号/状态/时间范围过滤，返回汇总+明细。"""
    result = {'success': False, 'message': ''}
    try:
        pk = params_kw or {}
        # 条件列统一加 u. 前缀（2026-09-10）：JOIN llm_model/llm_account/sd_projects
        # 后 org_id/status/created_at 多表同名，无前缀会触发 MariaDB 1052 歧义
        conds = ["1=1"]
        params = {}
        if pk.get('org_id'):
            conds.append("u.org_id=${o}$"); params['o'] = pk['org_id']
        if pk.get('model_id'):
            conds.append("u.model_id=${m}$"); params['m'] = pk['model_id']
        if pk.get('account_id'):
            conds.append("u.account_id=${a}$"); params['a'] = pk['account_id']
        if pk.get('status'):
            conds.append("u.status=${s}$"); params['s'] = pk['status']
        # 项目过滤（2026-09-10 用户定夺：按项目统计费用）；
        # 不传/null = 不区分项目（全部），'0' = 精确查非项目调用
        _pid = pk.get('project_id')
        if _pid:
            conds.append("u.project_id=${pid}$"); params['pid'] = _pid
        if pk.get('date_from'):
            conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from']
        if pk.get('date_to'):
            conds.append("u.created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59'
        where = " AND ".join(conds)
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                # token 数从 usages JSON 提取（req/resp_tokens 冗余列已删 m0021，
                # usages 是唯一事实源）；JSON_VALID 守卫防脏行报错；出参键名保持
                # req_tokens/resp_tokens 兼容既有前端
                "SELECT u.id, u.org_id, u.user_id, u.model_id, m.name AS model_name, "
                "u.account_id, a.name AS account_name, u.endpoint_region, "
                "CASE WHEN JSON_VALID(u.usages) THEN "
                "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') "
                "ELSE '0' END AS req_tokens, "
                "CASE WHEN JSON_VALID(u.usages) THEN "
                "COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') "
                "ELSE '0' END AS resp_tokens, "
                "u.cost, u.charge, u.ppid, u.project_id, "
                "CASE WHEN u.project_id='0' THEN '（非项目调用）' "
                "ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, "
                "u.task_ref, u.call_id, "
                "u.status, u.note, u.created_at "
                "FROM llm_usage u "
                "LEFT JOIN llm_model m ON m.id=u.model_id "
                "LEFT JOIN llm_account a ON a.id=u.account_id "
                "LEFT JOIN sd_projects sp ON sp.id=u.project_id "
                "WHERE %s ORDER BY u.created_at DESC LIMIT 200" % where, params)
            # token 汇总走 JSON_EXTRACT（列已删 m0021）：CAST DECIMAL 防字符串数字；
            # 非法/缺失 JSON 行按 0 计入，不中断汇总
            sums = await sor.sqlExe(
                "SELECT COUNT(*) AS cnt, "
                "COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN "
                "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.prompt_tokens')),'0') "
                "AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, "
                "COALESCE(SUM(CASE WHEN JSON_VALID(usages) THEN "
                "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(usages,'$.completion_tokens')),'0') "
                "AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t, "
                "COALESCE(SUM(cost),0) AS cost_s, "
                "COALESCE(SUM(charge),0) AS charge_s FROM llm_usage u WHERE %s" % where, params)
            await sor.sqlExe("COMMIT", {})
        rows = []
        for r in (recs or []):
            # sqlor 行是 DictObject（dict 子类），必须 dict() 拷贝——
            # vars() 取到的是空实例 __dict__（同 gateway._row_to_dict 根因）
            try:
                d = dict(r)
            except (TypeError, ValueError):
                d = {k: v for k, v in vars(r).items() if not callable(v)}
            rows.append(d)
        s = sums[0] if sums else None
        result['success'] = True
        result['rows'] = rows
        result['summary'] = {
            'count': int(getattr(s, 'cnt', 0)) if s else 0,
            'req_tokens': int(getattr(s, 'req_t', 0)) if s else 0,
            'resp_tokens': int(getattr(s, 'resp_t', 0)) if s else 0,
            'cost': round(float(getattr(s, 'cost_s', 0)), 6) if s else 0,
            'charge': round(float(getattr(s, 'charge_s', 0)), 6) if s else 0,
        }
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


async def llm_project_cost_query(params_kw):
    """按项目统计费用（2026-09-10 用户定夺）：llm_usage.project_id GROUP BY。

    过滤：org_id / date_from / date_to / status（缺省统计全部状态）。
    分组：project_id（'0' = 非项目调用哨兵值，一组显示'（非项目调用）'；
    其余 JOIN sd_projects 取项目名，项目已删显示'(项目已删除)'）。
    project_id 过滤：不传/null = 不区分项目（全部分组都出）。
    汇总口径：calls 调用次数 / cost 成本合计 / charge 客户应付合计 /
    req_t、resp_t token 合计（usages JSON 唯一事实源，JSON_VALID 守卫）。
    """
    result = {'success': False, 'message': ''}
    try:
        pk = params_kw or {}
        conds = ["1=1"]
        params = {}
        if pk.get('org_id'):
            conds.append("u.org_id=${o}$"); params['o'] = pk['org_id']
        if pk.get('status'):
            conds.append("u.status=${s}$"); params['s'] = pk['status']
        if pk.get('date_from'):
            conds.append("u.created_at>=${df}$"); params['df'] = pk['date_from']
        if pk.get('date_to'):
            conds.append("u.created_at<=${dt}$"); params['dt'] = pk['date_to'] + ' 23:59:59'
        where = " AND ".join(conds)
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            recs = await sor.sqlExe(
                "SELECT u.project_id, "
                "CASE WHEN u.project_id='0' THEN '（非项目调用）' "
                "ELSE COALESCE(sp.name, '(项目已删除)') END AS project_name, "
                "COUNT(*) AS calls, "
                "COALESCE(SUM(u.cost),0) AS cost_s, "
                "COALESCE(SUM(u.charge),0) AS charge_s, "
                "COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN "
                "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.prompt_tokens')),'0') "
                "AS DECIMAL(20,4)) ELSE 0 END),0) AS req_t, "
                "COALESCE(SUM(CASE WHEN JSON_VALID(u.usages) THEN "
                "CAST(COALESCE(JSON_UNQUOTE(JSON_EXTRACT(u.usages,'$.completion_tokens')),'0') "
                "AS DECIMAL(20,4)) ELSE 0 END),0) AS resp_t "
                "FROM llm_usage u "
                "LEFT JOIN sd_projects sp ON sp.id=u.project_id "
                "WHERE %s GROUP BY u.project_id, sp.name "
                "ORDER BY cost_s DESC, calls DESC LIMIT 500" % where, params)
            await sor.sqlExe("COMMIT", {})
        rows = []
        for r0 in (recs or []):
            try:
                d = dict(r0)
            except (TypeError, ValueError):
                d = {k: v for k, v in vars(r0).items() if not callable(v)}
            rows.append({
                'project_id': d.get('project_id'),
                'project_name': d.get('project_name') or '',
                'calls': int(d.get('calls', 0) or 0),
                'cost': round(float(d.get('cost_s', 0) or 0), 6),
                'charge': round(float(d.get('charge_s', 0) or 0), 6),
                'req_tokens': int(d.get('req_t', 0) or 0),
                'resp_tokens': int(d.get('resp_t', 0) or 0),
            })
        result['success'] = True
        result['rows'] = rows
        result['summary'] = {
            'project_count': len([x for x in rows
                                  if x['project_id'] and x['project_id'] != '0']),
            'calls': sum(x['calls'] for x in rows),
            'cost': round(sum(x['cost'] for x in rows), 6),
            'charge': round(sum(x['charge'] for x in rows), 6),
        }
    except Exception as e:
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


# ────────────────────── 调用原文追踪：「查看IO」弹窗 ──────────────────────

_IO_BODY_MAX = 150000   # 弹窗单侧最大展示字符（完整原文仍在盘上，超长提示文件路径）

_KIND_TEXT = {'chat': '对话', 'gen': '同步生成', 'submit': '异步提交', 'query': '任务轮询'}


def _read_trace_file(rel_path):
    """读 trace 原文文件。rel_path 为库中 filesroot 相对路径（llm_trace/<日>/<call>_<seq>.req.json）。

    安全铁律：路径来自数据库，realpath 后必须落在 llm_trace 目录内，逃逸=拒绝访问
    （防路径穿越；realpath 同时防符号链接指向目录外）。返回 (text, err)。
    """
    from .trace import _filesroot, trace_root
    if not rel_path:
        return '', '无原文文件（该侧落盘失败或已过保留期被清理）'
    root = os.path.realpath(trace_root())
    real = os.path.realpath(os.path.join(_filesroot(), str(rel_path)))
    if not real.startswith(root + os.sep):
        return '', '[非法路径，拒绝访问]'
    try:
        with open(real, 'r', encoding='utf-8') as f:
            return f.read(), ''
    except Exception as e:
        return '', '[文件读取失败: %s]' % str(e)[:120]


def _pretty_or_raw(text):
    """落盘 JSON 美化（indent=2）便于阅读；非合法 JSON 原样返回。"""
    try:
        return json.dumps(json.loads(text), ensure_ascii=False, indent=2)
    except Exception:
        return text


def _msg_widget(message):
    return {"widgettype": "Message",
            "options": {"title": "提示", "message": message,
                        "cwidth": 24, "cheight": 8, "timeout": 3}}


async def llm_call_trace_io(params_kw, user_org_id='0'):
    """「查看IO」弹窗内容组装：TabPanel（上下文摘要/请求原文/响应原文）。

    dspy 薄代理调本函数（读表+读文件+组装 widget 全在模块侧）。
    机构隔离：平台业主（org_id='0'）可看全部；普通机构只看本机构或 org_id 为空的行。
    返回 widget 描述 dict（外层 PopupWindow 壳由列表页 binds popup_options 生成，
    此处再包 PopupWindow 会套娃）。
    """
    pk = params_kw or {}
    trace_id = str(pk.get('id') or '').strip()
    if not trace_id or len(trace_id) > 32:
        return _msg_widget('缺少或非法的 trace id 参数')

    db, dbname = _get_sor()

    def _row2dict(r):
        # sqlExe 行可能是 DictObject(dict 子类) 或普通对象——与 llm_usage_query 同款双兜底
        try:
            return dict(r)
        except (TypeError, ValueError):
            return {k: v for k, v in vars(r).items() if not callable(v)}

    async with db.sqlorContext(dbname) as sor:
        recs = await sor.sqlExe(
            "SELECT * FROM llm_call_trace WHERE id=${id}$", {'id': trace_id})
        row = _row2dict(recs[0]) if recs else None
        batch_rows, usage_rows = [], []
        if row and row.get('call_id'):
            call_id = row['call_id']
            recs2 = await sor.sqlExe(
                "SELECT seq, kind, method, status_code, elapsed_ms, url, note "
                "FROM llm_call_trace WHERE call_id=${c}$ ORDER BY seq, created_at",
                {'c': call_id})
            batch_rows = [_row2dict(r) for r in (recs2 or [])]
            recs3 = await sor.sqlExe(
                "SELECT u.status, u.accounting_status, u.cost, u.charge, u.task_ref, "
                "u.project_id, m.name AS model_name "
                "FROM llm_usage u LEFT JOIN llm_model m ON m.id=u.model_id "
                "WHERE u.call_id=${c}$", {'c': call_id})
            usage_rows = [_row2dict(r) for r in (recs3 or [])]
        await sor.sqlExe("COMMIT", {})

    if not row:
        return _msg_widget('未找到调用追踪记录（可能已过保留期被清理）')
    row_org = str(row.get('org_id') or '')
    if str(user_org_id or '0') != '0' and row_org not in ('', str(user_org_id)):
        return _msg_widget('无权查看其他机构的调用原文')

    # 读文件在 DB 块外（行/批次/usage 已全部拷贝为本地 dict，不再依赖 sor）
    req_text, req_err = _read_trace_file(row.get('req_path') or '')
    resp_text, resp_err = _read_trace_file(row.get('resp_path') or '')
    if req_err:
        req_text = req_err
    else:
        req_text = _pretty_or_raw(req_text)
        if len(req_text) > _IO_BODY_MAX:
            req_text = req_text[:_IO_BODY_MAX] + \
                '\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('req_path') or '')
    if resp_err:
        resp_text = resp_err
    else:
        resp_text = _pretty_or_raw(resp_text)
        if len(resp_text) > _IO_BODY_MAX:
            resp_text = resp_text[:_IO_BODY_MAX] + \
                '\n\n...［内容过长已截断，完整原文见服务器文件: %s］' % (row.get('resp_path') or '')

    # ── 摘要 tab（本行 + 计费流水上下文 + 同批次往返）──
    lines = [
        '批次 call_id: %s' % (row.get('call_id') or ''),
        '往返 seq/类型: %s / %s' % (
            row.get('seq'), _KIND_TEXT.get(str(row.get('kind') or ''), row.get('kind') or '')),
        'HTTP: %s    状态码: %s    耗时: %s ms' % (
            row.get('method') or '', row.get('status_code'), row.get('elapsed_ms')),
        'URL: %s' % (row.get('url') or ''),
        '机构: %s' % (row_org or '(空)'),
        '时间: %s' % (row.get('created_at') or ''),
    ]
    if row.get('note'):
        lines.append('错误摘要: %s' % row['note'])
    if usage_rows:
        lines.append('')
        lines.append('—— 计费流水（llm_usage，同 call_id，%d 条）——' % len(usage_rows))
        for u in usage_rows:
            lines.append('模型: %s   状态: %s   记账: %s   成本: %s   应付: %s   来源: %s   项目: %s' % (
                u.get('model_name') or '(无)', u.get('status') or '',
                u.get('accounting_status') or '-', u.get('cost'), u.get('charge'),
                u.get('task_ref') or '-', u.get('project_id') or '-'))
    if batch_rows:
        lines.append('')
        lines.append('—— 同批次往返（%d 次）——' % len(batch_rows))
        for t in batch_rows:
            lines.append('seq=%s  %s  %s  状态:%s  %sms  %s%s' % (
                t.get('seq'),
                _KIND_TEXT.get(str(t.get('kind') or ''), t.get('kind') or ''),
                t.get('method') or '', t.get('status_code'), t.get('elapsed_ms'),
                (t.get('url') or '')[:100],
                ('  err: ' + str(t.get('note'))[:80]) if t.get('note') else ''))
    summary_text = '\n'.join(lines)

    # 摘要 tab 用 Html widget：Text 的 innerHTML 会吞换行且不转义（正文含上游
    # 响应片段时 < > 会破坏结构/注入），Html + <pre> + escape 是唯一安全渲染。
    summary_html = ('<pre style="white-space:pre-wrap;word-break:break-all;'
                    'font-size:13px;margin:0;font-family:inherit;">'
                    + html.escape(summary_text) + '</pre>')

    # mode:null → CodeMirror 纯文本+行号（仓库 3parties 只有 codemirror.js
    # 内建 null 模式和 codemirror-python.js；不指定 mode 会去拉不存在的
    # codemirror-python.js 之外的 mode 文件报 warn，null 是最干净的退化）。
    def _code_tab(name, label, content):
        return {"name": name, "label": label,
                "content": {"widgettype": "CodeEditor",
                            "options": {"value": content, "mode": "null",
                                        "readonly": True, "theme": False,
                                        "width": "100%", "height": "100%"}}}

    return {
        "widgettype": "TabPanel",
        "options": {
            "width": "100%", "height": "100%", "tab_pos": "top",
            "items": [
                {"name": "summary", "label": "上下文摘要",
                 "content": {"widgettype": "VBox",
                             "options": {"width": "100%", "height": "100%",
                                         "overflow": "auto", "padding": "12px"},
                             "subwidgets": [
                                 {"widgettype": "Html",
                                  "options": {"width": "100%", "height": "auto",
                                              "html": summary_html}}]}},
                _code_tab('req', '请求原文（上行）', req_text),
                _code_tab('resp', '响应原文（下行）', resp_text),
            ]
        }
    }


async def llm_dashboard(params_kw):
    """治理总览（首页卡片数据源）：各机构策略/余额/近期用量。"""
    result = {'success': True}
    try:
        db, dbname = _get_sor()
        async with db.sqlorContext(dbname) as sor:
            vendors = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_vendor WHERE status='active'", {})
            accounts = await sor.sqlExe(
                "SELECT COUNT(*) AS c, COALESCE(SUM(balance),0) AS b FROM llm_account WHERE status='active'", {})
            models = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_model WHERE status='active'", {})
            orgs = await sor.sqlExe("SELECT COUNT(*) AS c FROM llm_org_policy WHERE status='active'", {})
            usage = await sor.sqlExe(
                "SELECT COUNT(*) AS c, COALESCE(SUM(charge),0) AS ch FROM llm_usage WHERE status='SUCCEEDED'", {})
            await sor.sqlExe("COMMIT", {})
        result['vendors'] = int(getattr(vendors[0], 'c', 0)) if vendors else 0
        result['accounts'] = int(getattr(accounts[0], 'c', 0)) if accounts else 0
        result['account_balance'] = round(float(getattr(accounts[0], 'b', 0)), 4) if accounts else 0
        result['models'] = int(getattr(models[0], 'c', 0)) if models else 0
        result['org_policies'] = int(getattr(orgs[0], 'c', 0)) if orgs else 0
        result['usage_calls'] = int(getattr(usage[0], 'c', 0)) if usage else 0
        result['usage_charge'] = round(float(getattr(usage[0], 'ch', 0)), 4) if usage else 0
    except Exception as e:
        result['success'] = False
        result['message'] = str(e)
    return json.dumps(result, ensure_ascii=False, default=str)


def load_pipeline_llm():
    """模块加载：注册全部函数到 ServerEnv。（规范名，产品层按 load_{resource_module} 约定）"""
    load_gateway()
    load_inference()
    load_selection()
    env = ServerEnv()
    env.create_llm_vendor = create_llm_vendor
    env.update_llm_vendor = update_llm_vendor
    env.delete_llm_vendor = delete_llm_vendor
    env.create_llm_account = create_llm_account
    env.update_llm_account = update_llm_account
    env.delete_llm_account = delete_llm_account
    env.create_llm_model = create_llm_model
    env.update_llm_model = update_llm_model
    env.delete_llm_model = delete_llm_model
    env.update_llm_org_policy = update_llm_org_policy
    env.llm_usage_query = llm_usage_query
    env.llm_project_cost_query = llm_project_cost_query
    env.llm_call_trace_io = llm_call_trace_io
    env.llm_dashboard = llm_dashboard
    # 产品层标准接口（product_management.product_interface 调度）
    from .product_interface import (
        get_product_display, check_product_availability, check_product_consumable,
        execute_product_service, execute_product_service_stream,
        calculate_product_cost, load_product_category_product,
    )
    env.product_interface = {
        'module_name': 'pipeline_llm',
        'get_product_display': get_product_display,
        'check_product_availability': check_product_availability,
        'check_product_consumable': check_product_consumable,
        'execute_product_service': execute_product_service,
        'execute_product_service_stream': execute_product_service_stream,
        'calculate_product_cost': calculate_product_cost,
        'load_product_category_product': load_product_category_product,
    }
    env.load_product_category_product_llm = load_product_category_product
    # 平台待办 provider（软注册，2026-09-11：账号钱包余额不足 → 属主机构
    # operator 角色催充值待办；pipeline-service 没装/没钩子时只告警不崩，
    # 与 ticket 模块同款模式）
    try:
        from pipeline_service.human_task_capability import register_todo_provider
        from .todos import list_wallet_todos
        register_todo_provider(list_wallet_todos)
        logger.info("[pipeline_llm] wallet todo provider registered")
    except ImportError:
        logger.warning("[pipeline_llm] pipeline_service.register_todo_provider 不可用，"
                       "钱包余额待办不会出现在平台待办（模块可独立运行）")
    logger.info("[pipeline_llm] v1.0.0 loaded — 模型治理模块就绪")


# 兼容旧名（宿主曾用 load_llm；规范名为 load_pipeline_llm）
load_llm = load_pipeline_llm