fix: 上传/智能问答使用 Function key 时用环境变量做 Embedding 鉴权；新增 upload_to_kb 脚本

Made-with: Cursor

publish_with_profile.py

@@ -1,12 +1,5 @@
 #!/usr/bin/env python3
-"""
+"""使用 Azure 发布配置文件 (.PublishSettings) 将本项目部署到 Function App。"""
-使用 Azure 发布配置文件 (.PublishSettings) 将本项目部署到 Function App。
-无需 az login，只需从 Azure 门户下载的 agnetdoc.PublishSettings。
-
-用法:
-  python publish_with_profile.py [路径/to/agnetdoc.PublishSettings]
-  未传路径时默认当前目录下的 agnetdoc.PublishSettings
-"""
 import os
 import sys
 import zipfile
@@ -20,51 +13,42 @@ except ImportError:
     print("请先安装: pip install requests")
     sys.exit(1)
 
-# 打 zip 时排除的目录/文件
 EXCLUDE = {
-    ".venv",
+    ".venv", "venv", "__pycache__", ".git", ".azure", ".python_packages",
-    "venv",
+    "local.settings.json", "publish_with_profile.py", ".gitignore", "deploy.zip",
-    "__pycache__",
-    ".git",
-    ".azure",
-    ".python_packages",
-    "local.settings.json",
-    "publish_with_profile.py",
-    ".gitignore",
-    "deploy.zip",
 }
 
 
-def find_publish_profile(profile_path: Path) -> ET.Element:
+def find_zipdeploy_profile(profile_path: Path) -> ET.Element:
-    """从 XML 中取第一个 publishProfile。"""
+    """优先取 ZipDeploy 的 publishProfile。"""
     tree = ET.parse(profile_path)
-    root_el = tree.getroot()
+    for profile in tree.getroot().findall(".//publishProfile"):
-    for profile in root_el.findall(".//publishProfile"):
+        if (profile.get("publishMethod") or "").strip() == "ZipDeploy":
-        method = (profile.get("publishMethod") or "").strip()
-        if "MSDeploy" in method or "ZipDeploy" in method or method == "":
             return profile
-    first = root_el.find(".//publishProfile")
+    for profile in tree.getroot().findall(".//publishProfile"):
-    if first is not None:
+        if "MSDeploy" in (profile.get("publishMethod") or ""):
-        return first
+            return profile
-    for tag in ("publishProfile", "PublishProfile"):
+    first = tree.getroot().find(".//publishProfile")
-        for el in root_el.iter(tag):
+    return first
-            return el
-    return None
 
 
-def get_scm_url(destination_app_url: str) -> str:
+def get_scm_url(profile: ET.Element, destination_app_url: str) -> str:
-    """由 destinationAppUrl 得到 SCM (Kudu) 地址。"""
+    """从 profile 或 destinationAppUrl 得到 SCM 的 base URL。"""
+    # ZipDeploy 的 publishUrl 即 SCM 主机:port
+    pub_url = (profile.get("publishUrl") or "").strip()
+    if pub_url and "scm." in pub_url:
+        host_port = pub_url.split("/")[0]
+        return f"https://{host_port}"
     parsed = urlparse(destination_app_url)
     host = parsed.netloc
+    # 区域化: xxx.southeastasia-01.azurewebsites.net -> xxx.scm.southeastasia-01.azurewebsites.net
     if ".azurewebsites.net" in host and ".scm." not in host:
-        host = host.replace(".azurewebsites.net", ".scm.azurewebsites.net")
+        parts = host.split(".", 1)
-    else:
+        host = parts[0] + ".scm." + parts[1] if len(parts) == 2 else host
-        host = "agnetdoc.scm.azurewebsites.net"
     return f"{parsed.scheme}://{host}"
 
 
 def make_zip(project_dir: Path, zip_path: Path) -> None:
-    """把项目打成 zip，排除 EXCLUDE。"""
     with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
         for root, dirs, files in os.walk(project_dir):
             dirs[:] = [d for d in dirs if d not in EXCLUDE and not d.startswith(".")]
@@ -72,67 +56,45 @@ def make_zip(project_dir: Path, zip_path: Path) -> None:
             for f in files:
                 if f in EXCLUDE or f.endswith(".pyc"):
                     continue
-                path = Path(root) / f
+                zf.write(Path(root) / f, rel_root / f)
-                arc = rel_root / f
-                zf.write(path, arc)
 
 
 def main() -> None:
     script_dir = Path(__file__).resolve().parent
-    if len(sys.argv) >= 2:
+    profile_path = Path(sys.argv[1]) if len(sys.argv) >= 2 else script_dir / "agnetdoc.PublishSettings"
-        profile_path = Path(sys.argv[1])
-    else:
-        profile_path = script_dir / "agnetdoc.PublishSettings"
-
     if not profile_path.exists():
-        print(f"未找到发布配置文件: {profile_path}")
+        print(f"未找到: {profile_path}")
-        print("用法: python publish_with_profile.py <路径/to/agnetdoc.PublishSettings>")
         sys.exit(1)
-
     try:
-        profile = find_publish_profile(profile_path)
+        profile = find_zipdeploy_profile(profile_path)
         if profile is None:
-            print("发布配置文件中未找到 publishProfile 节点")
+            print("未找到 publishProfile")
             sys.exit(1)
         user = (profile.get("userName") or "").strip()
         pwd = (profile.get("userPWD") or "").strip()
         dest = profile.get("destinationAppUrl") or "https://agnetdoc.azurewebsites.net"
         if not user or not pwd:
-            print("发布配置中缺少 userName 或 userPWD")
+            print("发布配置中缺少 userName 或 userPWD，请从 Azure 门户重新下载发布配置文件")
             sys.exit(1)
     except ET.ParseError as e:
-        print(f"解析发布配置文件失败: {e}")
+        print(f"解析失败: {e}")
         sys.exit(1)
-
+    scm_url = get_scm_url(profile, dest)
-    scm_url = get_scm_url(dest)
     zip_deploy_url = f"{scm_url}/api/zipdeploy"
     zip_path = script_dir / "deploy.zip"
-
+    print("打包中...")
-    print("正在打包项目...")
     make_zip(script_dir, zip_path)
-    print(f"正在上传到 {scm_url} ...")
+    print(f"上传到 {scm_url} ...")
-
     try:
         with open(zip_path, "rb") as f:
-            r = requests.post(
+            r = requests.post(zip_deploy_url, auth=(user, pwd), data=f,
-                zip_deploy_url,
+                              headers={"Content-Type": "application/zip"}, timeout=600)
-                auth=(user, pwd),
-                data=f,
-                headers={"Content-Type": "application/zip"},
-                timeout=600,
-            )
     finally:
         zip_path.unlink(missing_ok=True)
-
     if r.status_code not in (200, 201, 202):
-        print(f"部署失败: HTTP {r.status_code}")
+        print(f"部署失败 HTTP {r.status_code}:", r.text[:400])
-        print(r.text[:500] if r.text else "")
         sys.exit(1)
-
+    print("部署已提交，可访问:", dest.rstrip("/"))
-    print("部署请求已接受，稍等片刻即可访问:")
-    app_url = dest.rstrip("/")
-    print(f"  {app_url}/")
-    print(f"  {app_url}/health")
 
 
 if __name__ == "__main__":

src/server/api_server.py

@@ -230,19 +230,25 @@ class SmartQueryRequest(BaseModel):
     top: int = Field(5, description="返回结果数量")
 
 
+def _use_request_key_for_llm(api_key: str) -> bool:
+    """请求中的 key 形如 sk-xxx 时才用于 LiteLLM/embedding，否则用环境变量中的 key。"""
+    return api_key and api_key.strip().startswith("sk-")
+
+
 @app.post("/api/v1/upload")
 async def api_upload(request: UploadRequest, api_key: str = Depends(verify_api_key)):
     """上传文档到资源库"""
     try:
         old_key = os.environ.get("OPENAI_API_KEY")
-        os.environ["OPENAI_API_KEY"] = api_key
+        if _use_request_key_for_llm(api_key):
+            os.environ["OPENAI_API_KEY"] = api_key.strip()
         try:
             result = await TOOL_MAP["upload_documents"](
                 documents=json.dumps(request.documents), index_name=request.index_name
             )
             return json.loads(result)
         finally:
-            if old_key:
+            if old_key is not None:
                 os.environ["OPENAI_API_KEY"] = old_key
     except Exception as e:
         raise HTTPException(status_code=500, detail=str(e))
@@ -317,7 +323,8 @@ async def api_smart_query(request: SmartQueryRequest, api_key: str = Depends(ver
     """AI 智能问答"""
     try:
         old_key = os.environ.get("OPENAI_API_KEY")
-        os.environ["OPENAI_API_KEY"] = api_key
+        if _use_request_key_for_llm(api_key):
+            os.environ["OPENAI_API_KEY"] = api_key.strip()
         try:
             result = await TOOL_MAP["smart_query"](
                 question=request.question,

upload_to_kb.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+""" one-off: 从 local.settings.json 加载环境变量并上传文档到知识库（Azure Search） """
+import asyncio
+import json
+import os
+import sys
+
+# 加载 local.settings.json
+settings_path = os.path.join(os.path.dirname(__file__), "local.settings.json")
+if not os.path.exists(settings_path):
+    print("未找到 local.settings.json", file=sys.stderr)
+    sys.exit(1)
+with open(settings_path, "r", encoding="utf-8") as f:
+    settings = json.load(f)
+for k, v in settings.get("Values", {}).items():
+    os.environ.setdefault(k, v)
+
+# 将项目根加入 path 以便 import src
+sys.path.insert(0, os.path.dirname(__file__))
+
+from src.server.mcp_server import upload_documents
+
+
+async def main():
+    doc_path = os.path.join(os.path.dirname(__file__), "../../../../ww/对接示例/API_DOCUMENTATION.md")
+    doc_path = os.path.normpath(doc_path)
+    if not os.path.exists(doc_path):
+        print(f"未找到文档: {doc_path}", file=sys.stderr)
+        sys.exit(1)
+    with open(doc_path, "r", encoding="utf-8") as f:
+        content = f.read()
+
+    doc = {
+        "id": "ww-api-doc-intelligent-search-agent",
+        "title": "智能搜索AI Agent API接口文档",
+        "content": content,
+        "project": "openclaw",
+        "category": "对接示例",
+        "tags": "API,智能搜索,Agent,对接",
+        "source": "human",
+        "author": "对接示例",
+    }
+    result = await upload_documents(documents=json.dumps([doc]), index_name="openclaw-resources")
+    print(result)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())