# -*- 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": "推荐理由", // 会自动清洗 引用标记 "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 清洗： 引用标记、顿号、「引用」、空括号。""" 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()