402 lines
18 KiB
Python
402 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
添加受众目标入库脚本（完整链路）。
|
||
|
||
链路（与前端 block-chat-audience.taskAddActionDesign / audience-content.taskActionDesignDel 一致）：
|
||
1. POST {login} 登录换取 access_token（Authorization: Bearer）
|
||
2. GET {design_list}?filter=taskId&filter=actionId 查询当前行动下已有受众（查重）
|
||
3. 同名受众存在时执行删除链：
|
||
GET {pathway_relation_list}?filter=actionId 找出该受众的 pathwayId（关系表）
|
||
DELETE {pathway_account_delete}/{pathwayId} 删关系表 task-action-pathway-account
|
||
DELETE {delete_design_vulns}?designId={id} 删受众关联脆弱点
|
||
DELETE {design_delete}/{id} 删受众目标本身
|
||
4. POST {design_create} 创建受众目标（parentTargetAudience=大模型生成）
|
||
5. POST {design_account_create} ×N 逐个创建受众-账号关联
|
||
|
||
用法：
|
||
python add_audiences.py --json payload.json [--dry-run] [--config ../config/audience_api.json]
|
||
|
||
payload.json 结构：
|
||
{
|
||
"taskId": 101,
|
||
"actionId": 201,
|
||
"audienceList": [
|
||
{ "targetAudience": "群体名", // 也可用 groupName
|
||
"recReason": "推荐理由", // 会自动清洗 <n> 引用标记
|
||
"userList": [ {完整user对象} ] } // 取 id/name/platformType/dataSource/score，整个对象进 frofileInfo
|
||
]
|
||
}
|
||
|
||
输出：一段 JSON（stdout），包含每个接口的入参与调用结果（密码与 token 已脱敏），
|
||
以及分组结果与汇总，供模型如实转述给用户。
|
||
"""
|
||
import argparse
|
||
import copy
|
||
import json
|
||
import os
|
||
import re
|
||
import ssl
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
DEFAULT_CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "config", "audience_api.json")
|
||
|
||
# 环境变量覆盖名
|
||
ENV_OVERRIDES = {
|
||
"base_url": "AUDIENCE_API_BASE_URL",
|
||
"username": "AUDIENCE_API_USERNAME",
|
||
"password": "AUDIENCE_API_PASSWORD",
|
||
"verify_ssl": "AUDIENCE_API_VERIFY_SSL",
|
||
}
|
||
|
||
|
||
def load_config(path: str) -> dict:
|
||
cfg = json.loads(open(path, encoding="utf-8").read())
|
||
for key, env in ENV_OVERRIDES.items():
|
||
if os.environ.get(env):
|
||
raw = os.environ[env]
|
||
if key == "verify_ssl":
|
||
cfg.setdefault("request", {})
|
||
cfg["request"]["verify_ssl"] = raw.lower() in ("1", "true", "yes")
|
||
else:
|
||
cfg[key] = raw
|
||
# 兼容 "$ENV_VAR" 占位值写法
|
||
def resolve(v):
|
||
if isinstance(v, str) and v.startswith("$") and re.fullmatch(r"\$[A-Z0-9_]+", v):
|
||
return os.environ.get(v[1:], "")
|
||
return v
|
||
|
||
def walk(o):
|
||
if isinstance(o, dict):
|
||
return {k: walk(v) for k, v in o.items()}
|
||
if isinstance(o, list):
|
||
return [walk(v) for v in o]
|
||
return resolve(o)
|
||
|
||
return walk(cfg)
|
||
|
||
|
||
def clean_reason(text) -> str:
|
||
"""复刻前端 recReason 清洗：<n> 引用标记、顿号、「引用」、空括号。"""
|
||
if not isinstance(text, str):
|
||
return ""
|
||
text = re.sub(r"<(\d+)>", "", text)
|
||
return text.replace("、", "").replace("引用", "").replace("()", "")
|
||
|
||
|
||
def platform_type_change(platform, options):
|
||
"""
|
||
平台名/别名 → 平台字典编码（写入 task-action-design-account.accountPlat）。
|
||
移植自前端 chat-data.utils.platformTypeChange：子串匹配 fb / x|witter / in / re / th → 字典 label
|
||
命中取 value；否则 label 全量小写精确匹配；仍匹配不到回退 1（facebook，与前端一致）。
|
||
options 形如 [{"label": "facebook", "value": 1}, ...]，来自 sys-dict-data 的 platform 字典。
|
||
技能侧扩展：入参已是编码（int 或纯数字字符串）时原样透传；完全缺失（None/空）返回 None（不臆造）。
|
||
"""
|
||
if platform is None or (isinstance(platform, str) and not platform.strip()):
|
||
return None
|
||
if isinstance(platform, bool):
|
||
return None
|
||
if isinstance(platform, int):
|
||
return platform
|
||
if isinstance(platform, str) and platform.strip().isdigit():
|
||
return int(platform.strip())
|
||
|
||
low = str(platform).strip().lower()
|
||
|
||
def val(*labels):
|
||
for lab in labels:
|
||
for o in options:
|
||
if str(o.get("label", "")).strip().lower() == lab and o.get("value") is not None:
|
||
return o.get("value")
|
||
return None
|
||
|
||
if "fb" in low:
|
||
return val("facebook") or 1
|
||
if "x" in low or "witter" in low:
|
||
return val("x(twitter)", "x") or 1
|
||
if "in" in low:
|
||
return val("instagram") or 1
|
||
if "re" in low:
|
||
return val("reddit") or 1
|
||
if "th" in low:
|
||
return val("threads") or 1
|
||
for o in options:
|
||
if str(o.get("label", "")).strip().lower() == low and o.get("value") is not None:
|
||
return o.get("value")
|
||
return 1
|
||
|
||
|
||
class ApiClient:
|
||
def __init__(self, cfg: dict, dry_run: bool = False):
|
||
self.base = cfg["base_url"].rstrip("/")
|
||
self.cfg = cfg
|
||
self.dry_run = dry_run
|
||
self.headers = {"Content-Type": "application/json"}
|
||
self.headers.update(cfg.get("request", {}).get("extra_headers") or {})
|
||
self.timeout = cfg.get("request", {}).get("timeout_seconds", 30)
|
||
verify = cfg.get("request", {}).get("verify_ssl", True)
|
||
self.ctx = ssl.create_default_context() if verify else ssl._create_unverified_context()
|
||
self.calls = []
|
||
|
||
def _log(self, step, method, url, request, status, response):
|
||
self.calls.append({
|
||
"step": step, "method": method, "url": url,
|
||
"request": request, "response": {"status": status, "body": response},
|
||
})
|
||
|
||
def request(self, step, method, path, query=None, body=None, auth=True):
|
||
"""执行一次 HTTP 请求并记录。返回 (status, 解析后的响应)。"""
|
||
url = self.base + path
|
||
if query:
|
||
url += "?" + urllib.parse.urlencode(query, doseq=True)
|
||
payload = json.dumps(body, ensure_ascii=False).encode("utf-8") if body is not None else None
|
||
display_body = self._redact(copy.deepcopy(body))
|
||
req_headers = dict(self.headers)
|
||
if auth:
|
||
req_headers[self.cfg["auth"]["header_name"]] = "***已附带鉴权头***"
|
||
|
||
if self.dry_run:
|
||
self._log(step, method, url, {"query": query, "body": display_body}, 0, "[dry-run] 未实际发送")
|
||
return 0, None
|
||
|
||
req = urllib.request.Request(url, data=payload, method=method)
|
||
for k, v in req_headers.items():
|
||
if v == "***已附带鉴权头***":
|
||
req.add_header(k, self._auth_header_value())
|
||
else:
|
||
req.add_header(k, v)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=self.timeout, context=self.ctx) as resp:
|
||
status, text = resp.status, resp.read().decode("utf-8", "replace")
|
||
except urllib.error.HTTPError as e:
|
||
status, text = e.code, e.read().decode("utf-8", "replace")
|
||
except Exception as e: # 网络层错误
|
||
self._log(step, method, url, {"query": query, "body": display_body}, -1, str(e))
|
||
raise
|
||
|
||
try:
|
||
parsed = json.loads(text) if text else None
|
||
except ValueError:
|
||
parsed = text[:300]
|
||
self._log(step, method, url, {"query": query, "body": display_body}, status, parsed if not isinstance(parsed, str) else parsed)
|
||
return status, parsed
|
||
|
||
def _auth_header_value(self) -> str:
|
||
return self.cfg["auth"]["header_template"].format(token=getattr(self, "token", ""))
|
||
|
||
@staticmethod
|
||
def _redact(body):
|
||
if isinstance(body, dict):
|
||
for k in list(body.keys()):
|
||
if k in ("password",):
|
||
body[k] = "***"
|
||
return body
|
||
|
||
|
||
def ep(cfg, key) -> str:
|
||
return cfg["endpoints"][key]
|
||
|
||
|
||
def main():
|
||
parser = argparse.ArgumentParser(description="添加受众目标（完整链路：登录→查重→删同名及关系表→建受众→建账号关联）")
|
||
parser.add_argument("--json", required=True, help="载荷 JSON 文件路径")
|
||
parser.add_argument("--config", default=DEFAULT_CONFIG, help="接口配置文件路径")
|
||
parser.add_argument("--dry-run", action="store_true", help="只校验与打印调用计划，不实际发送")
|
||
args = parser.parse_args()
|
||
|
||
result = {
|
||
"success": False, "dry_run": args.dry_run,
|
||
"配置": {}, "入参摘要": {}, "登录": None, "groups": [], "summary": {}, "error": None,
|
||
}
|
||
|
||
try:
|
||
cfg = load_config(args.config)
|
||
except Exception as e:
|
||
result["error"] = f"配置加载失败：{e}"
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
# ---- 载荷校验 ----
|
||
try:
|
||
payload = json.loads(open(args.json, encoding="utf-8").read())
|
||
except Exception as e:
|
||
result["error"] = f"载荷 JSON 读取失败：{e}"
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
task_id = payload.get("taskId")
|
||
action_id = payload.get("actionId")
|
||
groups_in = payload.get("audienceList") or []
|
||
errors = []
|
||
if not isinstance(task_id, int) or not isinstance(action_id, int):
|
||
errors.append("taskId / actionId 必须为整数")
|
||
if not groups_in:
|
||
errors.append("audienceList 不能为空")
|
||
for i, g in enumerate(groups_in):
|
||
name = (g.get("targetAudience") or g.get("groupName") or "").strip()
|
||
if not name:
|
||
errors.append(f"audienceList[{i}] 缺少群体名（targetAudience/groupName）")
|
||
if not isinstance(g.get("userList"), list):
|
||
errors.append(f"audienceList[{i}] 缺少 userList 数组")
|
||
if errors:
|
||
result["error"] = "载荷校验失败：" + "；".join(errors)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
result["配置"] = {
|
||
"base_url": cfg["base_url"], "username": cfg["username"],
|
||
"verify_ssl": cfg.get("request", {}).get("verify_ssl", True),
|
||
"timeout_seconds": cfg.get("request", {}).get("timeout_seconds", 30),
|
||
}
|
||
result["入参摘要"] = {
|
||
"taskId": task_id, "actionId": action_id,
|
||
"audienceList": [
|
||
{"targetAudience": (g.get("targetAudience") or g.get("groupName", "")).strip(),
|
||
"recReason清洗后": clean_reason(g.get("recReason", "")),
|
||
"账号数": len(g.get("userList") or [])}
|
||
for g in groups_in
|
||
],
|
||
}
|
||
|
||
client = ApiClient(cfg, dry_run=args.dry_run)
|
||
|
||
# ---- 1. 登录 ----
|
||
login_status, login_resp = client.request("登录", "POST", ep(cfg, "login"),
|
||
body={"username": cfg["username"], "password": cfg["password"]}, auth=False)
|
||
if login_status not in (200, 201):
|
||
result["登录"] = f"失败（HTTP {login_status}）"
|
||
result["error"] = f"登录失败：HTTP {login_status}，请检查 config/audience_api.json 的 base_url 与账号密码"
|
||
result["calls"] = client.calls
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
token = (login_resp or {}).get(cfg["auth"]["token_field"]) if isinstance(login_resp, dict) else None
|
||
if not token:
|
||
result["登录"] = "失败（响应中无 " + cfg["auth"]["token_field"] + "）"
|
||
result["error"] = f"登录响应缺少 {cfg['auth']['token_field']} 字段：{login_resp}"
|
||
result["calls"] = client.calls
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
client.token = token
|
||
result["登录"] = f"成功（已获取 {cfg['auth']['token_field']}，后续请求头 {cfg['auth']['header_name']}: {cfg['auth']['header_template']}）"
|
||
|
||
# ---- 1.5 平台字典（platform 名称/别名 → 字典编码，写入 accountPlat 前映射；与前端 platformTypeChange 同款） ----
|
||
platform_options = []
|
||
dict_status, dict_resp = client.request("查询平台字典", "GET", ep(cfg, "platform_dict"),
|
||
query={"filter": ["dictType||$eq||platform"], "limit": 99999})
|
||
dict_rows = (dict_resp or {}).get("data") or [] if isinstance(dict_resp, dict) else []
|
||
if dict_status == 200:
|
||
platform_options = [
|
||
{"label": r.get("dictLabel"), "value": int(str(r["dictValue"]).strip())}
|
||
for r in dict_rows
|
||
if r.get("dictValue") is not None and str(r.get("dictValue")).strip().lstrip("-").isdigit()
|
||
]
|
||
result["平台字典"] = (
|
||
f"已加载 {len(platform_options)} 项："
|
||
+ "、".join(f"{o['label']}={o['value']}" for o in platform_options)
|
||
)
|
||
else:
|
||
result["平台字典"] = f"查询失败（HTTP {dict_status}），平台名只能按子串规则映射且未命中回退 1，建议检查字典表"
|
||
|
||
# ---- 2. 查询已有受众（查重） ----
|
||
q = {"filter": [f"taskId||$eq||{task_id}", f"actionId||$eq||{action_id}"], "limit": 99999}
|
||
status, designs_resp = client.request("查询已有受众", "GET", ep(cfg, "design_list"), query=q)
|
||
designs = (designs_resp or {}).get("data") or [] if isinstance(designs_resp, dict) else []
|
||
if status != 200:
|
||
result["error"] = f"查询已有受众失败：HTTP {status}"
|
||
result["calls"] = client.calls
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(1)
|
||
|
||
# 关系表（一次查出整个行动的 pathway 关联）
|
||
rel_status, rel_resp = client.request("查询受众-传播节点关系表", "GET", ep(cfg, "pathway_relation_list"),
|
||
query={"filter": [f"actionId||$eq||{action_id}"], "limit": 99999})
|
||
relations = (rel_resp or {}).get("data") or [] if isinstance(rel_resp, dict) else []
|
||
|
||
summary = {"新增受众": 0, "新增账号关联": 0, "删除同名旧受众": 0, "删除关系表记录": 0, "删除脆弱点关联": 0}
|
||
all_ok = True
|
||
|
||
for g in groups_in:
|
||
name = (g.get("targetAudience") or g.get("groupName")).strip()
|
||
group = {"targetAudience": name, "designId": None, "新增账号": 0, "deletedOld": [], "ok": False, "error": None}
|
||
|
||
# ---- 3. 同名删除链 ----
|
||
for old in [d for d in designs if str(d.get("targetAudience", "")).strip() == name]:
|
||
old_id = old.get("id")
|
||
pathway_ids = [r.get("pathwayId") for r in relations if r.get("designId") == old_id and r.get("pathwayId") is not None]
|
||
for pid in pathway_ids:
|
||
s, _ = client.request("删除关系表记录", "DELETE", f"{ep(cfg, 'pathway_account_delete')}/{pid}")
|
||
if s in (200, 204):
|
||
summary["删除关系表记录"] += 1
|
||
s, _ = client.request("删除受众关联脆弱点", "DELETE", ep(cfg, "delete_design_vulns"),
|
||
query={"designId": old_id})
|
||
if s in (200, 204):
|
||
summary["删除脆弱点关联"] += 1
|
||
s, _ = client.request("删除同名旧受众", "DELETE", f"{ep(cfg, 'design_delete')}/{old_id}")
|
||
if s in (200, 204):
|
||
summary["删除同名旧受众"] += 1
|
||
group["deletedOld"].append({"id": old_id, "删除关系表记录数": len(pathway_ids)})
|
||
|
||
# ---- 4. 创建受众目标 ----
|
||
design_body = {
|
||
"taskId": task_id, "actionId": action_id,
|
||
"parentTargetAudience": "大模型生成",
|
||
"targetAudience": name,
|
||
"recReason": clean_reason(g.get("recReason", "")),
|
||
}
|
||
s, created = client.request("创建受众目标", "POST", ep(cfg, "design_create"), body=design_body)
|
||
if s not in (200, 201) or not isinstance(created, dict) or "id" not in created:
|
||
group["error"] = f"创建受众目标失败：HTTP {s}，响应：{created}"
|
||
result["groups"].append(group)
|
||
all_ok = False
|
||
continue
|
||
design_id = created["id"]
|
||
group["designId"] = design_id
|
||
summary["新增受众"] += 1
|
||
|
||
# ---- 5. 创建账号关联 ----
|
||
for user in g.get("userList") or []:
|
||
account_source = user.get("dataSource") if user.get("dataSource") is not None else user.get("score")
|
||
# 平台编码：user.platformType（编码）缺失时回退 user.platform（名称），统一经平台字典映射为编码
|
||
raw_platform = user.get("platformType") if user.get("platformType") is not None else user.get("platform")
|
||
account_plat = platform_type_change(raw_platform, platform_options)
|
||
group.setdefault("账号平台映射", []).append(
|
||
{"accountName": user.get("name"), "原始值": raw_platform, "accountPlat": account_plat}
|
||
)
|
||
account_body = {
|
||
"taskId": task_id, "actionId": action_id,
|
||
"accountId": user.get("id"),
|
||
"accountName": user.get("name"),
|
||
"accountPlat": account_plat,
|
||
"accountSource": account_source,
|
||
"frofileInfo": json.dumps(user, ensure_ascii=False),
|
||
"designId": design_id,
|
||
}
|
||
s, resp = client.request("创建受众-账号关联", "POST", ep(cfg, "design_account_create"), body=account_body)
|
||
if s in (200, 201):
|
||
group["新增账号"] += 1
|
||
summary["新增账号关联"] += 1
|
||
else:
|
||
group.setdefault("账号失败", []).append({"accountName": user.get("name"), "status": s, "response": resp})
|
||
all_ok = False
|
||
|
||
group["ok"] = "账号失败" not in group
|
||
result["groups"].append(group)
|
||
|
||
result["summary"] = summary
|
||
result["success"] = all_ok
|
||
result["calls"] = client.calls
|
||
result["人读摘要"] = (
|
||
f"{'[演练]' if args.dry_run else ''}共处理 {len(groups_in)} 个受众组："
|
||
f"新增受众 {summary['新增受众']} 个、账号关联 {summary['新增账号关联']} 条；"
|
||
f"删除同名旧受众 {summary['删除同名旧受众']} 个（连带关系表记录 {summary['删除关系表记录']} 条、脆弱点关联 {summary['删除脆弱点关联']} 处）。"
|
||
+ ("全部成功。" if all_ok else "存在失败项，详见 groups/error 字段。")
|
||
)
|
||
print(json.dumps(result, ensure_ascii=False, indent=2))
|
||
sys.exit(0 if all_ok else 2)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|