zw/iOSAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

20250918-新增主播库功能

Browse Source
This commit is contained in:
zhangkai
2025-09-18 20:15:07 +08:00
parent 2fe1576eaa
commit 11e72d0fae
6 changed files with 327 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

177
Utils/AiUtils.py
View File

@@ -614,22 +614,55 @@ class AiUtils(object):
with open(file_path, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False, indent=2)
# @staticmethod
# def _normalize_anchor_items(items):
# """
# 规范化输入为 [{anchorId, country}] 的列表：
# - 允许传入：单个对象、对象列表、字符串（当 anchorId 用）
# - 过滤不合规项
# """
# result = []
# if items is None:
# return result
#
# if isinstance(items, dict):
# # 单个对象
# aid = items.get("anchorId")
# if aid:
# result.append({"anchorId": str(aid), "country": items.get("country", "")})
# return result
#
# if isinstance(items, list):
# for it in items:
# if isinstance(it, dict):
# aid = it.get("anchorId")
# if aid:
# result.append({"anchorId": str(aid), "country": it.get("country", "")})
# elif isinstance(it, str):
# result.append({"anchorId": it, "country": ""})
# return result
#
# if isinstance(items, str):
# result.append({"anchorId": items, "country": ""})
# return result
@staticmethod
def _normalize_anchor_items(items):
"""
规范化输入为 [{anchorId, country}] 的列表：
- 允许传入：单个对象、对象列表、字符串（当 anchorId 用）
- 过滤不合规项
规范化为 [{...}]：
- 允许传入：单个对象、对象数组、字符串（当 anchorId 用）
- 保留原有字段，不限制只存 anchorId/country
- 字符串输入 → {"anchorId": xxx}
"""
result = []
if items is None:
return result
if isinstance(items, dict):
# 单个对象
aid = items.get("anchorId")
if aid:
result.append({"anchorId": str(aid), "country": items.get("country", "")})
obj = dict(items) # 保留所有字段
result.append(obj)
return result
if isinstance(items, list):
@@ -637,15 +670,17 @@ class AiUtils(object):
if isinstance(it, dict):
aid = it.get("anchorId")
if aid:
result.append({"anchorId": str(aid), "country": it.get("country", "")})
obj = dict(it)
result.append(obj)
elif isinstance(it, str):
result.append({"anchorId": it, "country": ""})
result.append({"anchorId": it})
return result
if isinstance(items, str):
result.append({"anchorId": items, "country": ""})
result.append({"anchorId": items})
return result
# -------- 追加（对象数组平铺追加） --------
@classmethod
def save_aclist_flat_append(cls, acList, filename="log/acList.json"):
@@ -657,6 +692,8 @@ class AiUtils(object):
{"anchorId": "tianliang30", "country": ""}
]
"""
file_path = Path(filename)
data = cls._read_json_list(file_path)
@@ -690,6 +727,97 @@ class AiUtils(object):
cls._write_json_list(file_path, data)
return first
@classmethod
def bulk_update_anchors(cls, updates, filename="log/acList.json", case_insensitive=False):
"""
批量修改（文件根必须是数组，沿用 _read_json_list 的约定）
- updates:
dict: {"id1": {...}, "id2": {...}}
list[dict]: [{"anchorId":"id1", ...}, {"anchorId":"id2", ...}]
- case_insensitive: True 时用小写比较 anchorId
返回: {"updated": <int>, "missing": [ids...], "file": "<实际命中的路径>"}
"""
def norm_id(x: str) -> str:
s = str(x).strip()
return s.lower() if case_insensitive else s
# ✅ 关键：使用你已有的 _resolve_path，避免受 cwd 影响
file_path = cls._resolve_path(filename)
data = cls._read_json_list(file_path)
if not data:
return {"updated": 0, "missing": cls._collect_all_ids(updates), "file": str(file_path)}
# 1) 归一化 updates -> map[normalized_id] = patch
upd_map = {}
raw_ids = [] # 保留原始传入 id，用于返回 missing 时回显
if isinstance(updates, dict):
for aid, patch in updates.items():
if aid and isinstance(patch, dict):
key = norm_id(aid)
raw_ids.append(str(aid))
patch = {k: v for k, v in patch.items() if k != "anchorId"}
if patch:
upd_map[key] = {**upd_map.get(key, {}), **patch}
elif isinstance(updates, list):
for it in updates:
if isinstance(it, dict) and it.get("anchorId"):
rid = str(it["anchorId"])
key = norm_id(rid)
raw_ids.append(rid)
patch = {k: v for k, v in it.items() if k != "anchorId"}
if patch:
upd_map[key] = {**upd_map.get(key, {}), **patch}
if not upd_map:
return {"updated": 0, "missing": [], "file": str(file_path)}
# 2) 建索引：map[normalized_id] -> item
index = {}
for item in data:
if isinstance(item, dict) and "anchorId" in item:
key = norm_id(item.get("anchorId", ""))
if key:
index[key] = item
# 3) 执行更新
updated, seen = 0, set()
for key, patch in upd_map.items():
target = index.get(key)
if target is not None:
target.update(patch)
updated += 1
seen.add(key)
# 4) 写回
if updated > 0:
cls._write_json_list(file_path, data)
# 5) 计算未命中（按传入原始 ID 回显）
missing = []
for rid in raw_ids:
if norm_id(rid) not in seen:
missing.append(rid)
return {"updated": updated, "missing": missing, "file": str(file_path)}
@staticmethod
def _collect_all_ids(updates):
ids = []
if isinstance(updates, dict):
ids = [str(k) for k in updates.keys()]
elif isinstance(updates, list):
for it in updates:
if isinstance(it, dict) and it.get("anchorId"):
ids.append(str(it["anchorId"]))
return ids
@classmethod
def delete_anchors_by_ids(cls, ids: list[str], filename="log/acList.json") -> int:
"""
@@ -718,6 +846,39 @@ class AiUtils(object):
LogManager.error(f"[delete_anchors_by_ids] 写入失败: {e}")
return deleted
# -------- 查看第一个（取出但不删除） --------
@staticmethod
def _resolve_path(p) -> Path:
p = Path(p)
if p.is_absolute():
return p
# 以项目根目录 (iOSAI) 为基准 —— script 的上一级
base = Path(__file__).resolve().parents[1]
return (base / p).resolve()
@classmethod
def peek_aclist_first(cls, filename="Module/log/acList.json"):
file_path = cls._resolve_path(filename)
if not file_path.exists():
print(f"[peek] 文件不存在: {file_path}")
return None
try:
raw = file_path.read_text(encoding="utf-8-sig").strip()
if not raw:
return None
data = json.loads(raw)
arr = data if isinstance(data, list) else data.get("anchorList") if isinstance(data, dict) else None
if not arr:
return None
first = arr[0]
norm = cls._normalize_anchor_items(first)
return norm[0] if norm else None
except Exception as e:
print(f"[peek] 读取失败: {e}")
return None
@staticmethod
def run_tidevice_command(udid, action, bundle_id, timeout=30):
"""