309 lines
13 KiB
Python
309 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
在 Azure AI 搜索中枚举并配置：数据源、技能组、索引器。
|
|
技能组是 AI 搜索内置功能，内置技能（如 OCR）在技能组内声明，由索引器执行。
|
|
创建技能组时若未配置 COGNITIVE_SERVICES_KEY，会通过 az login 枚举 Azure 认知服务/内容识别并自动读取终结点与密钥。
|
|
"""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
|
|
# 加载 local.settings.json
|
|
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
|
|
SETTINGS_PATH = os.path.join(SCRIPT_DIR, "local.settings.json")
|
|
if os.path.exists(SETTINGS_PATH):
|
|
with open(SETTINGS_PATH, "r", encoding="utf-8") as f:
|
|
for k, v in json.load(f).get("Values", {}).items():
|
|
os.environ.setdefault(k, v)
|
|
|
|
ENDPOINT = os.environ.get("AZURE_SEARCH_ENDPOINT", "").rstrip("/")
|
|
API_KEY = os.environ.get("AZURE_SEARCH_API_KEY", "")
|
|
API_VERSION = "2024-07-01"
|
|
INDEXER_API_VERSION = "2025-09-01" # 索引器创建需较新版本
|
|
|
|
|
|
def _headers():
|
|
return {"Content-Type": "application/json", "api-key": API_KEY}
|
|
|
|
|
|
def _get(path):
|
|
import urllib.request
|
|
req = urllib.request.Request(f"{ENDPOINT}{path}", headers=_headers(), method="GET")
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def _put(path, body):
|
|
import urllib.request
|
|
data = json.dumps(body).encode("utf-8")
|
|
req = urllib.request.Request(f"{ENDPOINT}{path}", data=data, headers=_headers(), method="PUT")
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode()) if resp.length else {}
|
|
|
|
|
|
def _post(path, body):
|
|
import urllib.request
|
|
data = json.dumps(body).encode("utf-8")
|
|
req = urllib.request.Request(f"{ENDPOINT}{path}", data=data, headers=_headers(), method="POST")
|
|
with urllib.request.urlopen(req) as resp:
|
|
return json.loads(resp.read().decode()) if resp.length else {}
|
|
|
|
|
|
def get_cognitive_services_from_az():
|
|
"""
|
|
通过 az login 枚举当前订阅下的认知服务/内容识别（Microsoft.CognitiveServices/accounts），
|
|
并读取每个资源的终结点与 key。返回列表 [{name, resourceGroup, endpoint, key}, ...]。
|
|
若 az 未登录或未安装，返回 []。
|
|
"""
|
|
out = []
|
|
try:
|
|
# 枚举认知服务账户（含多服务/内容识别/Foundry）
|
|
r = subprocess.run(
|
|
["az", "cognitiveservices", "account", "list", "-o", "json"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=30,
|
|
)
|
|
if r.returncode != 0 or not r.stdout.strip():
|
|
return out
|
|
accounts = json.loads(r.stdout)
|
|
if not isinstance(accounts, list):
|
|
return out
|
|
for acc in accounts:
|
|
name = acc.get("name") or ""
|
|
rg = acc.get("resourceGroup") or ""
|
|
# 终结点可能在 properties.endpoint 或 properties.endpoint 里
|
|
props = acc.get("properties") or {}
|
|
endpoint = (props.get("endpoint") or "").rstrip("/")
|
|
if not name or not rg:
|
|
continue
|
|
# 获取 key
|
|
kr = subprocess.run(
|
|
["az", "cognitiveservices", "account", "keys", "list", "--name", name, "--resource-group", rg, "-o", "json"],
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=15,
|
|
)
|
|
key = ""
|
|
if kr.returncode == 0 and kr.stdout.strip():
|
|
keys_obj = json.loads(kr.stdout)
|
|
key = (keys_obj.get("key1") or keys_obj.get("key2") or "").strip()
|
|
out.append({"name": name, "resourceGroup": rg, "endpoint": endpoint, "key": key})
|
|
return out
|
|
except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired) as e:
|
|
return out
|
|
|
|
|
|
def list_cognitive_services():
|
|
"""枚举并打印 Azure 中的认知服务/内容识别资源（终结点与密钥占位），便于确认。"""
|
|
print("=== Azure 认知服务 / 内容识别 (az login 枚举) ===\n")
|
|
items = get_cognitive_services_from_az()
|
|
if not items:
|
|
print("未发现认知服务资源，请确认：1) 已执行 az login 2) 当前订阅下已创建认知服务/内容识别。")
|
|
return
|
|
for i, x in enumerate(items, 1):
|
|
key_preview = (x["key"][:8] + "...") if x.get("key") else "(未读取)"
|
|
print(f" [{i}] {x['name']}")
|
|
print(f" 资源组: {x['resourceGroup']}")
|
|
print(f" 终结点: {x.get('endpoint') or '(空)'}")
|
|
print(f" Key: {key_preview}")
|
|
print()
|
|
print("创建技能组时将自动使用上述第一个资源的 key（或通过 COGNITIVE_SERVICES_KEY 指定）。")
|
|
|
|
|
|
def enumerate_services():
|
|
"""枚举当前 Search 服务中的：数据源、技能组、索引器、索引"""
|
|
if not ENDPOINT or not API_KEY:
|
|
print("缺少 AZURE_SEARCH_ENDPOINT 或 AZURE_SEARCH_API_KEY（请在 local.settings.json 中配置）")
|
|
return
|
|
base = f"?api-version={API_VERSION}"
|
|
print("=== 数据源 (Data Sources) ===")
|
|
try:
|
|
r = _get(f"/datasources{base}")
|
|
for x in r.get("value", []):
|
|
print(f" - {x.get('name')} type={x.get('type')}")
|
|
if not r.get("value"):
|
|
print(" (无)")
|
|
except Exception as e:
|
|
print(f" 错误: {e}")
|
|
print("\n=== 技能组 (Skillsets) ===")
|
|
try:
|
|
r = _get(f"/skillsets{base}")
|
|
for x in r.get("value", []):
|
|
print(f" - {x.get('name')} skills={len(x.get('skills', []))}")
|
|
if not r.get("value"):
|
|
print(" (无)")
|
|
except Exception as e:
|
|
print(f" 错误: {e}")
|
|
print("\n=== 索引器 (Indexers) ===")
|
|
try:
|
|
r = _get(f"/indexers{base}")
|
|
for x in r.get("value", []):
|
|
print(f" - {x.get('name')} dataSource={x.get('dataSourceName')} skillset={x.get('skillsetName')} index={x.get('targetIndexName')}")
|
|
if not r.get("value"):
|
|
print(" (无)")
|
|
except Exception as e:
|
|
print(f" 错误: {e}")
|
|
print("\n=== 索引 (Indexes) ===")
|
|
try:
|
|
r = _get(f"/indexes{base}")
|
|
for x in r.get("value", []):
|
|
print(f" - {x.get('name')}")
|
|
if not r.get("value"):
|
|
print(" (无)")
|
|
except Exception as e:
|
|
print(f" 错误: {e}")
|
|
|
|
|
|
def create_datasource_skillset_indexer():
|
|
"""
|
|
创建：Blob 数据源、技能组（OCR + Merge + 挂载 Azure 内容识别）、
|
|
用于 Blob 的索引、索引器。全部在 AI 搜索内完成。
|
|
"""
|
|
conn_str = os.environ.get("INDEXER_BLOB_CONNECTION_STRING") or os.environ.get("BLOB_STORAGE_CONNECTION_STRING") or os.environ.get("AzureWebJobsStorage")
|
|
container = os.environ.get("INDEXER_BLOB_CONTAINER", "documents")
|
|
cognitive_key = os.environ.get("COGNITIVE_SERVICES_KEY") or os.environ.get("CONTENT_SAFETY_KEY")
|
|
index_name = os.environ.get("INDEXER_TARGET_INDEX", "openclaw-blob")
|
|
|
|
if not ENDPOINT or not API_KEY:
|
|
print("缺少 AZURE_SEARCH_ENDPOINT 或 AZURE_SEARCH_API_KEY")
|
|
return False
|
|
if not conn_str:
|
|
print("缺少 Blob 连接字符串，请设置 INDEXER_BLOB_CONNECTION_STRING 或 BLOB_STORAGE_CONNECTION_STRING")
|
|
return False
|
|
# 未配置 key 时，通过 az 枚举认知服务/内容识别并自动读取 key
|
|
if not cognitive_key:
|
|
candidates = get_cognitive_services_from_az()
|
|
# 优先名称含「内容识别」或 content / cognitive 的资源
|
|
for c in candidates:
|
|
if c.get("key") and ("内容识别" in (c.get("name") or "") or "content" in (c.get("name") or "").lower() or "cognitive" in (c.get("name") or "").lower() or "foundry" in (c.get("name") or "").lower()):
|
|
cognitive_key = c["key"]
|
|
print(f"已从 Azure 读取认知服务: {c['name']} (资源组: {c['resourceGroup']})，终结点: {c.get('endpoint') or '(无)'}")
|
|
break
|
|
if not cognitive_key and candidates:
|
|
c = candidates[0]
|
|
if c.get("key"):
|
|
cognitive_key = c["key"]
|
|
print(f"已从 Azure 读取认知服务: {c['name']} (资源组: {c['resourceGroup']})，终结点: {c.get('endpoint') or '(无)'}")
|
|
if not cognitive_key:
|
|
print("未配置 COGNITIVE_SERVICES_KEY 且 az 未枚举到可用认知服务，请先 az login 并确保订阅下有认知服务/内容识别，或手动设置 COGNITIVE_SERVICES_KEY")
|
|
return False
|
|
|
|
base = f"?api-version={API_VERSION}"
|
|
|
|
# 1. 数据源
|
|
ds_name = "openclaw-blob-datasource"
|
|
ds_body = {
|
|
"name": ds_name,
|
|
"type": "azureblob",
|
|
"credentials": {"connectionString": conn_str},
|
|
"container": {"name": container},
|
|
}
|
|
try:
|
|
_put(f"/datasources('{ds_name}'){base}", ds_body)
|
|
print(f"已创建/更新数据源: {ds_name} container={container}")
|
|
except Exception as e:
|
|
print(f"创建数据源失败: {e}")
|
|
return False
|
|
|
|
# 2. 技能组（仅内置技能：OCR、Merge；计费挂载 Azure 内容识别）
|
|
ss_name = "openclaw-blob-skillset"
|
|
skillset_body = {
|
|
"name": ss_name,
|
|
"skills": [
|
|
{
|
|
"@odata.type": "#Microsoft.Skills.Vision.OcrSkill",
|
|
"context": "/document/normalized_images/*",
|
|
"defaultLanguageCode": None,
|
|
"detectOrientation": True,
|
|
"inputs": [{"name": "image", "source": "/document/normalized_images/*"}],
|
|
"outputs": [{"name": "text", "targetName": "ocr_text"}],
|
|
},
|
|
{
|
|
"@odata.type": "#Microsoft.Skills.Text.MergeSkill",
|
|
"context": "/document",
|
|
"insertPreTag": " ",
|
|
"insertPostTag": " ",
|
|
"inputs": [
|
|
{"name": "text", "source": "/document/content"},
|
|
{"name": "itemsToInsert", "source": "/document/normalized_images/*/ocr_text"},
|
|
{"name": "offsets", "source": "/document/normalized_images/*/contentOffset"},
|
|
],
|
|
"outputs": [{"name": "mergedText", "targetName": "merged_text"}],
|
|
},
|
|
],
|
|
"cognitiveServices": {
|
|
"@odata.type": "#Microsoft.Azure.Search.CognitiveServicesByKey",
|
|
"description": "Azure 内容识别，用于 OCR 等内置技能计费",
|
|
"key": cognitive_key,
|
|
},
|
|
}
|
|
try:
|
|
_put(f"/skillsets('{ss_name}'){base}", skillset_body)
|
|
print(f"已创建/更新技能组: {ss_name} (OCR + Merge，已挂载 cognitiveServices)")
|
|
except Exception as e:
|
|
print(f"创建技能组失败: {e}")
|
|
return False
|
|
|
|
# 3. 索引（供 Blob 管道写入）
|
|
index_body = {
|
|
"name": index_name,
|
|
"fields": [
|
|
{"name": "id", "type": "Edm.String", "key": True, "filterable": True},
|
|
{"name": "content", "type": "Edm.String", "searchable": True},
|
|
{"name": "metadata_storage_name", "type": "Edm.String", "filterable": True, "sortable": True},
|
|
{"name": "metadata_storage_path", "type": "Edm.String", "filterable": True},
|
|
{"name": "metadata_storage_last_modified", "type": "Edm.DateTimeOffset", "filterable": True, "sortable": True},
|
|
],
|
|
}
|
|
try:
|
|
_put(f"/indexes('{index_name}'){base}", index_body)
|
|
print(f"已创建/更新索引: {index_name}")
|
|
except Exception as e:
|
|
print(f"创建索引失败: {e}")
|
|
return False
|
|
|
|
# 4. 索引器（拉取 Blob → 执行技能组 → 写入索引）
|
|
idx_name = "openclaw-blob-indexer"
|
|
indexer_body = {
|
|
"name": idx_name,
|
|
"dataSourceName": ds_name,
|
|
"targetIndexName": index_name,
|
|
"skillsetName": ss_name,
|
|
"parameters": {
|
|
"configuration": {
|
|
"dataToExtract": "contentAndMetadata",
|
|
"imageAction": "generateNormalizedImages",
|
|
"parsingMode": "default",
|
|
},
|
|
},
|
|
"fieldMappings": [
|
|
{"sourceFieldName": "metadata_storage_path", "targetFieldName": "id"},
|
|
{"sourceFieldName": "metadata_storage_name", "targetFieldName": "metadata_storage_name"},
|
|
{"sourceFieldName": "metadata_storage_path", "targetFieldName": "metadata_storage_path"},
|
|
{"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "metadata_storage_last_modified"},
|
|
],
|
|
"outputFieldMappings": [
|
|
{"sourceFieldName": "/document/merged_text", "targetFieldName": "content"},
|
|
],
|
|
}
|
|
try:
|
|
_put(f"/indexers('{idx_name}')?api-version={INDEXER_API_VERSION}", indexer_body)
|
|
print(f"已创建/更新索引器: {idx_name}")
|
|
except Exception as e:
|
|
print(f"创建索引器失败: {e}")
|
|
return False
|
|
|
|
print("\n配置完成。可将文件放入 Blob 容器后运行索引器。")
|
|
return True
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) > 1 and sys.argv[1] == "--create":
|
|
create_datasource_skillset_indexer()
|
|
elif len(sys.argv) > 1 and sys.argv[1] == "--list-cognitive":
|
|
list_cognitive_services()
|
|
else:
|
|
enumerate_services()
|