137 lines
5.5 KiB
Python
137 lines
5.5 KiB
Python
"""百炼 AI 应用连通性测试 — 完整响应输出版本。
|
||
|
||
将 8 个应用的完整返回内容写入 export/bailian_test_results.txt。
|
||
"""
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
import urllib.error
|
||
from pathlib import Path
|
||
|
||
from dotenv import load_dotenv
|
||
|
||
# 加载 .env
|
||
load_dotenv(Path(__file__).resolve().parents[2] / ".env")
|
||
|
||
BAILIAN_API_KEY = os.environ.get("BAILIAN_API_KEY", "")
|
||
if not BAILIAN_API_KEY:
|
||
print("❌ BAILIAN_API_KEY 未设置"); sys.exit(1)
|
||
|
||
API_BASE = "https://dashscope.aliyuncs.com/api/v1/apps"
|
||
|
||
APP_CONFIGS = [
|
||
("BAILIAN_APP_ID_1_CHAT", "应用1-通用对话"),
|
||
("BAILIAN_APP_ID_2_FINANCE", "应用2-财务洞察"),
|
||
("BAILIAN_APP_ID_3_CLUE", "应用3-客户数据维客线索分析"),
|
||
("BAILIAN_APP_ID_4_ANALYSIS", "应用4-关系分析/任务建议"),
|
||
("BAILIAN_APP_ID_5_TACTICS", "应用5-话术参考"),
|
||
("BAILIAN_APP_ID_6_NOTE", "应用6-备注分析"),
|
||
("BAILIAN_APP_ID_7_CUSTOMER", "应用7-客户分析"),
|
||
("BAILIAN_APP_ID_8_CONSOLIDATE", "应用8-维客线索整理"),
|
||
]
|
||
|
||
# 简短示例 prompt（复用原脚本的首条 prompt 结构）
|
||
FIRST_PROMPTS = {
|
||
"BAILIAN_APP_ID_1_CHAT": json.dumps({
|
||
"page": "finance_dashboard",
|
||
"data": {"monthly_revenue": 12580, "revenue_change_pct": 8.3,
|
||
"breakdown": {"台费": 6200, "助教服务费": 3100, "商品": 1800, "充值": 1480}},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_2_FINANCE": json.dumps({
|
||
"page": "finance_dashboard",
|
||
"data": {"monthly_revenue": 12580, "top_expense": "人工成本 8200元"},
|
||
"user_query": "本月利润率怎么样？"
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_3_CLUE": json.dumps({
|
||
"customer": {"name": "张三", "total_visits": 12, "last_visit_days_ago": 13,
|
||
"avg_visit_interval_days": 10, "monthly_spend": 580},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_4_ANALYSIS": json.dumps({
|
||
"customer": {"name": "张三", "total_visits": 12, "last_visit_days_ago": 13,
|
||
"tags": ["高频", "台费为主"], "recent_trend": "消费频次下降"},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_5_TACTICS": json.dumps({
|
||
"scenario": "客户3天未到店，需要召回话术",
|
||
"customer": {"name": "张三", "preference": "周末晚间", "favorite_table": "9号台"},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_6_NOTE": json.dumps({
|
||
"notes": ["客户说最近比较忙", "上次来提到想学斯诺克", "朋友推荐来的"],
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_7_CUSTOMER": json.dumps({
|
||
"customer": {"name": "张三", "level": "VIP", "total_spend": 15800,
|
||
"visit_count": 45, "first_visit": "2025-06-15", "tags": ["高频", "储值用户"]},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
"BAILIAN_APP_ID_8_CONSOLIDATE": json.dumps({
|
||
"customer": {"name": "张三", "total_visits": 12, "last_visit_days_ago": 13,
|
||
"avg_visit_interval_days": 10, "monthly_spend": 580,
|
||
"notes": ["最近比较忙", "想学斯诺克"]},
|
||
"user_query": ""
|
||
}, ensure_ascii=False),
|
||
}
|
||
|
||
|
||
def call_app(env_key: str, label: str) -> dict:
|
||
app_id = os.environ.get(env_key, "")
|
||
if not app_id:
|
||
return {"app": label, "status": "SKIP", "text": f"环境变量 {env_key} 未设置"}
|
||
|
||
prompt = FIRST_PROMPTS.get(env_key, "你好")
|
||
url = f"{API_BASE}/{app_id}/completion"
|
||
body = json.dumps({"input": {"prompt": prompt}, "parameters": {"has_thoughts": False}}).encode("utf-8")
|
||
headers = {"Authorization": f"Bearer {BAILIAN_API_KEY}", "Content-Type": "application/json"}
|
||
|
||
t0 = time.time()
|
||
try:
|
||
req = urllib.request.Request(url, data=body, headers=headers, method="POST")
|
||
with urllib.request.urlopen(req, timeout=120) as resp:
|
||
data = json.loads(resp.read().decode("utf-8"))
|
||
elapsed = round(time.time() - t0, 2)
|
||
text = data.get("output", {}).get("text", "")
|
||
usage = data.get("usage", {})
|
||
return {"app": label, "status": "OK", "elapsed": elapsed, "text": text, "usage": usage}
|
||
except Exception as e:
|
||
elapsed = round(time.time() - t0, 2)
|
||
return {"app": label, "status": "ERROR", "elapsed": elapsed, "text": str(e)[:500]}
|
||
|
||
|
||
def main():
|
||
output_path = os.environ.get("EXPORT_ROOT")
|
||
if not output_path:
|
||
raise RuntimeError("EXPORT_ROOT 环境变量未设置")
|
||
out_file = Path(output_path) / "bailian_test_results.txt"
|
||
out_file.parent.mkdir(parents=True, exist_ok=True)
|
||
|
||
lines = []
|
||
lines.append("=" * 70)
|
||
lines.append("百炼 AI 应用连通性测试 — 完整响应")
|
||
lines.append(f"时间: {time.strftime('%Y-%m-%d %H:%M:%S')}")
|
||
lines.append("=" * 70)
|
||
|
||
for env_key, label in APP_CONFIGS:
|
||
print(f"▶ {label} ...", flush=True)
|
||
r = call_app(env_key, label)
|
||
icon = "✅" if r["status"] == "OK" else "❌"
|
||
print(f" {icon} {r['status']} ({r.get('elapsed', '?')}s)")
|
||
|
||
lines.append(f"\n{'─' * 70}")
|
||
lines.append(f"【{r['app']}】 {r['status']} ({r.get('elapsed', '?')}s)")
|
||
lines.append(f"{'─' * 70}")
|
||
lines.append(r["text"])
|
||
if "usage" in r:
|
||
lines.append(f"\n[usage] {json.dumps(r['usage'], ensure_ascii=False)}")
|
||
|
||
out_file.write_text("\n".join(lines), encoding="utf-8")
|
||
print(f"\n✅ 完整结果已写入: {out_file}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|