519 lines
23 KiB
Python
519 lines
23 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-11-01-Preview" # 索引器需此版本以支持 allowSkillsetToReadFileData
|
||
SKILLSET_API_VERSION = "2025-11-01-Preview" # 内容理解技能需此版本
|
||
|
||
|
||
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
|
||
import urllib.error
|
||
data = json.dumps(body).encode("utf-8")
|
||
req = urllib.request.Request(f"{ENDPOINT}{path}", data=data, headers=_headers(), method="PUT")
|
||
try:
|
||
with urllib.request.urlopen(req) as resp:
|
||
return json.loads(resp.read().decode()) if getattr(resp, "length", None) else {}
|
||
except urllib.error.HTTPError as e:
|
||
body = e.read().decode() if e.fp else ""
|
||
raise RuntimeError(f"HTTP {e.code}: {body}") from e
|
||
|
||
|
||
def _delete(path):
|
||
import urllib.request
|
||
req = urllib.request.Request(f"{ENDPOINT}{path}", headers=_headers(), method="DELETE")
|
||
with urllib.request.urlopen(req) as resp:
|
||
pass
|
||
|
||
|
||
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_ai_foundry_endpoint_from_az():
|
||
"""通过 ARM 获取 AIServices 资源的终结点。返回 URL 或 None。"""
|
||
ep, _ = get_aiservices_endpoint_and_key_from_az()
|
||
return ep
|
||
|
||
|
||
def get_aiservices_endpoint_and_key_from_az():
|
||
"""
|
||
通过 ARM 获取当前订阅下 AIServices 资源的终结点与 key（用于内容理解技能）。
|
||
仅查找 kind=AIServices 的资源（FormRecognizer 如 taijiagnetdocs 不支持内容理解技能）。
|
||
返回 (endpoint_url 或 None, key 或 None)。
|
||
"""
|
||
try:
|
||
r = subprocess.run(
|
||
["az", "account", "show", "-o", "json"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=10,
|
||
)
|
||
if r.returncode != 0 or not r.stdout.strip():
|
||
return None, None
|
||
sub = json.loads(r.stdout)
|
||
sid = sub.get("id") or ""
|
||
if not sid:
|
||
return None, None
|
||
subscription_id = sid if sid and "/" not in sid else (sid or "").split("/")[-1]
|
||
accs = subprocess.run(
|
||
["az", "cognitiveservices", "account", "list", "-o", "json"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=30,
|
||
)
|
||
if accs.returncode != 0 or not accs.stdout.strip():
|
||
return None, None
|
||
for acc in json.loads(accs.stdout) or []:
|
||
name, rg = acc.get("name") or "", acc.get("resourceGroup") or ""
|
||
if not name or not rg:
|
||
continue
|
||
url = (
|
||
f"https://management.azure.com/subscriptions/{subscription_id}"
|
||
f"/resourceGroups/{rg}/providers/Microsoft.CognitiveServices/accounts/{name}?api-version=2023-10-01-preview"
|
||
)
|
||
rest = subprocess.run(
|
||
["az", "rest", "--method", "get", "--url", url, "-o", "json"],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=15,
|
||
)
|
||
if rest.returncode != 0 or not rest.stdout.strip():
|
||
continue
|
||
data = json.loads(rest.stdout) or {}
|
||
if data.get("kind") != "AIServices":
|
||
continue
|
||
props = data.get("properties") or {}
|
||
endpoints = props.get("endpoints") or {}
|
||
foundry = (endpoints.get("AI Foundry API") or endpoints.get("Content Understanding") or "").rstrip("/")
|
||
if not foundry:
|
||
continue
|
||
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():
|
||
ko = json.loads(kr.stdout)
|
||
key = (ko.get("key1") or ko.get("key2") or "").strip()
|
||
return foundry, key or None
|
||
return None, None
|
||
except (FileNotFoundError, json.JSONDecodeError, subprocess.TimeoutExpired, KeyError):
|
||
return None, None
|
||
|
||
|
||
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")
|
||
cognitive_endpoint = os.environ.get("COGNITIVE_SERVICES_ENDPOINT", "").rstrip("/")
|
||
# 统一只用一个索引：API 上传与索引器（Blob/图片/音视频 OCR 等）都写入同一索引
|
||
index_name = os.environ.get("AZURE_SEARCH_INDEX_NAME") or os.environ.get("INDEXER_TARGET_INDEX", "openclaw-resources")
|
||
|
||
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
|
||
# 内容理解技能仅支持 AIServices 资源。从 Azure 读取 AIServices 的终结点与 key；若 env 指向的是 FormRecognizer（如 taijiagnetdocs）则忽略，改用 az 的 AIServices
|
||
cognitive_endpoint = None
|
||
_env_ep = os.environ.get("COGNITIVE_SERVICES_ENDPOINT", "").rstrip("/")
|
||
# 先尝试从 az 取 AIServices 资源（taijiagnet 等）的 Content Understanding / AI Foundry 终结点
|
||
cognitive_endpoint, cognitive_key_from_az = get_aiservices_endpoint_and_key_from_az()
|
||
if cognitive_endpoint:
|
||
if cognitive_key_from_az:
|
||
cognitive_key = cognitive_key_from_az
|
||
print(f"已从 Azure 读取 AIServices 资源终结点（内容理解）: {cognitive_endpoint}")
|
||
if not cognitive_endpoint and _env_ep and (".cognitiveservices.azure.com" in _env_ep or ".services.ai.azure.com" in _env_ep):
|
||
cognitive_endpoint = _env_ep
|
||
print(f"使用配置的终结点: {cognitive_endpoint}")
|
||
if not cognitive_endpoint:
|
||
cognitive_endpoint = _env_ep
|
||
if not cognitive_endpoint:
|
||
print(
|
||
"AIServicesByKey 需要 subdomainUrl（必须是 AI Services/Foundry 资源专属 URL）。\n"
|
||
"请设置 COGNITIVE_SERVICES_ENDPOINT 为 Foundry 资源的终结点，例如：\n"
|
||
" https://<资源名>.cognitiveservices.azure.com 或 https://<资源名>.services.ai.azure.com\n"
|
||
"（不能使用区域端点如 southeastasia.api.cognitive.microsoft.com）"
|
||
)
|
||
return False
|
||
|
||
base = f"?api-version={API_VERSION}"
|
||
|
||
# 统一命名（不再用 openclaw-blob-*，只保留一套：数据源、技能组、索引器 → 同一索引）
|
||
ds_name = "openclaw-datasource"
|
||
ss_name = "openclaw-skillset"
|
||
idx_name = "openclaw-indexer"
|
||
|
||
# 1. 数据源
|
||
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（保证索引器可运行）
|
||
use_content_understanding = True
|
||
func_url = (os.environ.get("AZURE_FUNCTION_APP_URL") or "").rstrip("/")
|
||
skill_safe_id = (
|
||
{
|
||
"@odata.type": "#Microsoft.Skills.Custom.WebApiSkill",
|
||
"uri": f"{func_url}/api/skill/safe-id",
|
||
"httpMethod": "POST",
|
||
"context": "/document",
|
||
"inputs": [{"name": "path", "source": "/document/metadata_storage_path"}],
|
||
"outputs": [{"name": "safe_id", "targetName": "safe_id"}],
|
||
}
|
||
if func_url
|
||
else None
|
||
)
|
||
skillset_body_cu = {
|
||
"name": ss_name,
|
||
"skills": (
|
||
([skill_safe_id] if skill_safe_id else [])
|
||
+ [
|
||
{
|
||
"@odata.type": "#Microsoft.Skills.Util.ContentUnderstandingSkill",
|
||
"description": "使用 Azure 内容理解分析文档、图像等，提取文本块与图片",
|
||
"context": "/document",
|
||
"extractionOptions": ["images", "locationMetadata"],
|
||
"chunkingProperties": {"unit": "characters", "maximumLength": 4000, "overlapLength": 200},
|
||
"inputs": [{"name": "file_data", "source": "/document/file_data"}],
|
||
"outputs": [
|
||
{"name": "text_sections", "targetName": "text_sections"},
|
||
{"name": "normalized_images", "targetName": "normalized_images"},
|
||
],
|
||
},
|
||
]
|
||
),
|
||
"cognitiveServices": {
|
||
"@odata.type": "#Microsoft.Azure.Search.AIServicesByKey",
|
||
"description": "Foundry 资源，内容理解技能仅支持 AIServicesByKey",
|
||
"key": cognitive_key,
|
||
"subdomainUrl": cognitive_endpoint,
|
||
},
|
||
}
|
||
skillset_body_ocr = {
|
||
"name": ss_name,
|
||
"skills": (
|
||
([skill_safe_id] if skill_safe_id else [])
|
||
+ [
|
||
{
|
||
"@odata.type": "#Microsoft.Skills.Vision.OcrSkill",
|
||
"context": "/document/normalized_images/*",
|
||
"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,
|
||
},
|
||
}
|
||
# 上面 OCR 的 cognitiveServices 在回退时可能被覆盖为 COGNITIVE_SERVICES_KEY_OCR
|
||
try:
|
||
_put(f"/skillsets('{ss_name}')?api-version={SKILLSET_API_VERSION}", skillset_body_cu)
|
||
print(f"已创建/更新技能组: {ss_name} (Azure 内容理解 ContentUnderstandingSkill)")
|
||
except Exception as e:
|
||
err = str(e)
|
||
if any(x in err for x in ("AIServices", "SubdomainUrl", "No such host", "InvalidApiType", "Unsupported Api Type")):
|
||
print(f"内容理解技能组创建失败 ({err[:90]}...)，回退为 OCR+Merge 技能组")
|
||
use_content_understanding = False
|
||
# OCR 技能需多服务/Foundry 类型 key；若当前 key 为智能文档专用，用 COGNITIVE_SERVICES_KEY_OCR
|
||
ocr_key = os.environ.get("COGNITIVE_SERVICES_KEY_OCR") or cognitive_key
|
||
skillset_body_ocr["cognitiveServices"] = {
|
||
"@odata.type": "#Microsoft.Azure.Search.CognitiveServicesByKey",
|
||
"description": "Azure 内容识别，OCR 计费",
|
||
"key": ocr_key,
|
||
}
|
||
try:
|
||
_put(f"/skillsets('{ss_name}'){base}", skillset_body_ocr)
|
||
print(f"已创建/更新技能组: {ss_name} (OCR + Merge)")
|
||
except Exception as e2:
|
||
print(f"创建技能组失败: {e2}")
|
||
return False
|
||
else:
|
||
print(f"创建技能组失败: {e}")
|
||
return False
|
||
|
||
# 3. 不单独建索引，使用已有索引 openclaw-resources
|
||
|
||
# 4. 索引器（按技能组类型选择参数与输出映射）
|
||
if use_content_understanding:
|
||
cu_field_mappings = [
|
||
{"sourceFieldName": "metadata_storage_name", "targetFieldName": "title"},
|
||
{"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "created_at"},
|
||
{"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "updated_at"},
|
||
]
|
||
cu_output_mappings = [
|
||
{"sourceFieldName": "/document/text_sections/0/content", "targetFieldName": "content"},
|
||
]
|
||
if skill_safe_id:
|
||
cu_output_mappings.insert(0, {"sourceFieldName": "/document/safe_id", "targetFieldName": "id"})
|
||
else:
|
||
cu_field_mappings.insert(0, {"sourceFieldName": "metadata_storage_path", "targetFieldName": "id"})
|
||
indexer_body = {
|
||
"name": idx_name,
|
||
"dataSourceName": ds_name,
|
||
"targetIndexName": index_name,
|
||
"skillsetName": ss_name,
|
||
"parameters": {
|
||
"configuration": {
|
||
"dataToExtract": "contentAndMetadata",
|
||
"parsingMode": "default",
|
||
"allowSkillsetToReadFileData": True,
|
||
},
|
||
},
|
||
"fieldMappings": cu_field_mappings,
|
||
"outputFieldMappings": cu_output_mappings,
|
||
}
|
||
else:
|
||
# 若启用了 safe-id 技能，id 由技能输出映射；否则仍用 metadata_storage_path（路径含中文会失败）
|
||
ocr_field_mappings = [
|
||
{"sourceFieldName": "metadata_storage_name", "targetFieldName": "title"},
|
||
{"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "created_at"},
|
||
{"sourceFieldName": "metadata_storage_last_modified", "targetFieldName": "updated_at"},
|
||
]
|
||
if not skill_safe_id:
|
||
ocr_field_mappings.insert(0, {"sourceFieldName": "metadata_storage_path", "targetFieldName": "id"})
|
||
ocr_output_mappings = [{"sourceFieldName": "/document/merged_text", "targetFieldName": "content"}]
|
||
if skill_safe_id:
|
||
ocr_output_mappings.insert(0, {"sourceFieldName": "/document/safe_id", "targetFieldName": "id"})
|
||
indexer_body = {
|
||
"name": idx_name,
|
||
"dataSourceName": ds_name,
|
||
"targetIndexName": index_name,
|
||
"skillsetName": ss_name,
|
||
"parameters": {
|
||
"configuration": {
|
||
"dataToExtract": "contentAndMetadata",
|
||
"imageAction": "generateNormalizedImages",
|
||
"parsingMode": "default",
|
||
},
|
||
},
|
||
"fieldMappings": ocr_field_mappings,
|
||
"outputFieldMappings": ocr_output_mappings,
|
||
}
|
||
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
|
||
|
||
# 5. 删除旧版命名资源（openclaw-blob-*），只保留一套
|
||
for old_name, kind in [
|
||
("openclaw-blob-indexer", "indexer"),
|
||
("openclaw-blob-skillset", "skillset"),
|
||
("openclaw-blob-datasource", "datasource"),
|
||
]:
|
||
try:
|
||
if kind == "indexer":
|
||
_delete(f"/indexers('{old_name}')?api-version={INDEXER_API_VERSION}")
|
||
elif kind == "skillset":
|
||
_delete(f"/skillsets('{old_name}')?api-version={SKILLSET_API_VERSION}")
|
||
else:
|
||
_delete(f"/{kind}s('{old_name}'){base}")
|
||
print(f"已删除旧资源: {old_name}")
|
||
except Exception as e:
|
||
if "404" in str(e) or "Not Found" in str(e):
|
||
pass # 本就不存在
|
||
else:
|
||
print(f"删除 {old_name} 时: {e}")
|
||
|
||
print(f"\n配置完成。仅保留一个索引: {index_name}；数据源/技能组/索引器已统一命名并指向该索引。")
|
||
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()
|