""" AIGC接口调用 调用AIGC接口创建爬取计划，绑定生成计划 """ import json import logging from datetime import datetime from typing import List, Dict, Union, Tuple, Any import requests from agent import ToolResult, tool logger = logging.getLogger(__name__) AIGC_BASE_URL = "https://aigc-api.aiddit.com" CRAWLER_PLAN_CREATE_URL = f"{AIGC_BASE_URL}/aigc/crawler/plan/save" GET_PRODUCE_PLAN_DETAIL_BY_ID = f"{AIGC_BASE_URL}/aigc/produce/plan/detail" PRODUCE_PLAN_SAVE = f"{AIGC_BASE_URL}/aigc/produce/plan/save" DEFAULT_TOKEN = "8bf14f27fc3a486788f3383452422d72" DEFAULT_TIMEOUT = 60.0 @tool(description="根据抖音账号ID创建爬取计划") async def create_crawler_plan_by_douyin_account_id( account_id: str, sort_type: str = "最新", produce_plan_ids: List[str] = [] ) -> ToolResult: """ 根据抖音账号ID创建爬取计划 Args: account_id: 抖音账号ID sort_type: 搜索时的视频排序方式（最新/最热），默认最新 produce_plan_ids: 爬取计划要绑定的生成计划ID，默认为空列表 Returns: ToolResult: 包含以下内容 - output: 文本格式的爬取计划创建结果摘要 - metadata.result: 结构化的爬取计划创建结果 - crawler_info: 爬取计划信息 - crawler_plan_id: 创建的爬取计划ID - crawler_plan_name: 创建的爬取计划名称 - sort_type: 排序方式 - produce_plan_infos: 绑定的生成计划信息 - produce_plan_id: 生成计划ID - produce_plan_name: 生成计划名称 - is_success: 是否成功, true表示绑定成功，false表示绑定失败 - msg: 绑定失败时为错误信息，绑定成功则为“成功” Note: - 建议从 metadata.result 获取结构化数据，而非解析 output 文本 """ # 验证 account_id 格式 if not account_id or not isinstance(account_id, str): logger.error(f"create_crawler_plan_by_douyin_account_id invalid account_id: {account_id}") return ToolResult( title="根据抖音账号ID创建爬取计划失败", output="", error="account_id 参数无效：必须是非空字符串", ) if not account_id.startswith("MS4wLjABAAAA"): logger.error(f"create_crawler_plan_by_douyin_account_id invalid sec_uid format account_id:{account_id}") return ToolResult( title="根据抖音账号ID创建爬取计划失败", output="", error=f"account_id 格式错误：必须以 MS4wLjABAAAA 开头，当前值: {account_id[:min(20, len(account_id))]}...", ) if produce_plan_ids is None: produce_plan_ids = [] dt = datetime.now().strftime("%Y%m%d%h%M%s") crawler_plan_name = f"【内容寻找Agent自动创建】{dt}_抖音账号ID爬取计划_{account_id[:min(30, len(account_id))]}" params = { "accountFilters": [], "channel": 2, "contentFilters": [], "contentModal": 4, "crawlerComment": 0, "crawlerMode": 4, "filterAccountMatchMode": 2, "filterContentMatchMode": 2, "frequencyType": 1, "inputModeValues": [ account_id ], "modelValueConfig": { "sortType": sort_type }, "name": crawler_plan_name, "planType": 2, "searchModeValues": [], "selectModeValues": [], "srtExtractFlag": 1, "videoKeyFrameType": 1, "voiceExtractFlag": 1 } try: summary_lines = [f"抖音账号【{account_id}】创建爬取计划"] response_json = post(CRAWLER_PLAN_CREATE_URL, params) if response_json.get("code") != 0: return ToolResult( title="根据抖音账号ID创建爬取计划失败", output=response_json.get("msg", "接口异常"), error=f"create crawler plan interface error", ) crawler_plan_id = response_json.get("data", {}).get("id", "") summary_lines.append(f"爬取计划名称: {crawler_plan_name}") summary_lines.append(f" 抖音账号ID: {account_id}") summary_lines.append(f" 爬取计划ID: {crawler_plan_id}") summary_lines.append(f" 爬取计划排序方式: {sort_type}") produce_plan_infos: List[Dict[str, str]] = [] if produce_plan_ids: input_source_info = { "contentType": 1, "inputSourceType": 2, "inputSourceValue": crawler_plan_id, "inputSourceLabel": f"原始帖子-视频-抖音-内容添加计划-{crawler_plan_name}", "inputSourceModal": 4, "inputSourceChannel": 2 } produce_plan_infos, msg = crawler_plan_bind_produce_plan(input_source_info, produce_plan_ids) if produce_plan_infos: for produce_plan_info in produce_plan_infos: summary_lines.append(" 绑定的生成计划列表: ") summary_lines.append(f" 生成计划名称: {produce_plan_info.get('produce_plan_name', '')}") summary_lines.append(f" 生成计划ID: {produce_plan_info.get('produce_plan_id', '')}") summary_lines.append(f" 绑定结果: {"绑定成功" if not produce_plan_info.get("msg") else "绑定失败"}") summary_lines.append(f" 信息: {produce_plan_info.get('msg', '成功')}") return ToolResult( title="根据抖音账号ID创建爬取计划", output="\n".join(summary_lines), metadata={ "result": { "crawler_info": { "crawler_plan_id": crawler_plan_id, "crawler_plan_name": crawler_plan_name, "sort_type": sort_type, }, "produce_plan_infos": [ { "produce_plan_id": produce_plan_info.get("produce_plan_id", ""), "produce_plan_name": produce_plan_info.get("produce_plan_name", ""), "is_success": "绑定成功" if not produce_plan_info.get("msg") else "绑定失败", "msg": produce_plan_info.get("msg", "成功"), } for produce_plan_info in produce_plan_infos ] } }, long_term_memory="Create crawler plan by DouYin Account ID", ) except Exception as e: logger.error(f"create douyin account crawler plan error: {str(e)}, account_id: {account_id} ") return ToolResult( title="根据抖音账号ID创建爬取计划失败", output="", error=f"创建爬取计划错误:{str(e)}", ) @tool(description="根据抖音视频ID创建爬取计划") async def create_crawler_plan_by_douyin_content_id( content_ids: List[str], produce_plan_ids: List[str] = [] ) -> ToolResult: """ 根据抖音视频ID创建爬取计划 Args: content_ids: 抖音内容ID列表 produce_plan_ids: 爬取计划要绑定的生成计划ID，默认为空列表 Returns: Returns: ToolResult: 包含以下内容 - output: 文本格式的爬取计划创建结果摘要 - metadata.result: 结构化的爬取计划创建结果 - crawler_info: 爬取计划信息 - crawler_plan_id: 创建的爬取计划ID - crawler_plan_name: 创建的爬取计划名称 - content_ids: 抖音视频ID列表 - produce_plan_infos: 绑定的生成计划信息 - produce_plan_id: 生成计划ID - produce_plan_name: 生成计划名称 - is_success: 是否成功, true表示绑定成功，false表示绑定失败 - msg: 绑定失败时为错误信息，绑定成功则为“成功” Note: - 建议从 metadata.result 获取结构化数据，而非解析 output 文本 """ if not content_ids or not isinstance(content_ids, list): logger.error(f"create_crawler_plan_by_douyin_content_id invalid content_ids. content_ids: {content_ids}") return ToolResult( title="根据抖音内容ID创建爬取计划失败", output="", error="content_ids 参数无效: content_ids必须是列表" ) if len(content_ids) > 100: logger.error(f"create_crawler_plan_by_douyin_content_id invalid content_ids length. content_ids.length: {len(content_ids)}") return ToolResult( title="根据抖音内容ID创建爬取计划失败", output="", error=f"content_ids 长度异常: 期望1~100, 实际{len(content_ids)}" ) dt = datetime.now().strftime("%Y%m%d%h%M%s") crawler_plan_name = f"【内容寻找Agent自动创建】抖音视频直接抓取-{dt}-抖音" params = { "channel": 2, "contentModal": 4, "crawlerComment": 0, "crawlerMode": 5, "filterAccountMatchMode": 2, "filterContentMatchMode": 2, "frequencyType": 2, "inputModeValues": content_ids, "name": crawler_plan_name, "planType": 2, "searchModeValues": [], "srtExtractFlag": 1, "videoKeyFrameType": 1, "voiceExtractFlag": 1 } try: summary_lines = [f"抖音视频爬取计划"] response_json = post(CRAWLER_PLAN_CREATE_URL, params) if response_json.get("code") != 0: return ToolResult( title="根据抖音内容ID创建爬取计划失败", output=response_json.get("msg", "接口异常"), error=f"create crawler plan interface error", ) crawler_plan_id = response_json.get("data", {}).get("id", "") summary_lines.append(f"爬取计划名称: {crawler_plan_name}") summary_lines.append(f" 抖音视频IDs: {','.join(content_ids)}") summary_lines.append(f" 爬取计划ID: {crawler_plan_id}") produce_plan_infos: List[Dict[str, str]] = [] if produce_plan_ids: input_source_info = { "contentType": 1, "inputSourceType": 2, "inputSourceValue": crawler_plan_id, "inputSourceLabel": f"原始帖子-视频-抖音-内容添加计划-{crawler_plan_name}", "inputSourceModal": 4, "inputSourceChannel": 2 } produce_plan_infos, msg = crawler_plan_bind_produce_plan(input_source_info, produce_plan_ids) if produce_plan_infos: for produce_plan_info in produce_plan_infos: summary_lines.append(" 绑定的生成计划列表: ") summary_lines.append(f" 生成计划名称: {produce_plan_info.get('produce_plan_name', '')}") summary_lines.append(f" 生成计划ID: {produce_plan_info.get('produce_plan_id', '')}") summary_lines.append(f" 绑定结果: {"绑定成功" if not produce_plan_info.get("msg") else "绑定失败"}") summary_lines.append(f" 信息: {produce_plan_info.get('msg', '成功')}") return ToolResult( title="根据抖音内容ID创建爬取计划", output="\n".join(summary_lines), metadata={ "result": { "crawler_info": { "crawler_plan_id": crawler_plan_id, "crawler_plan_name": crawler_plan_name, }, "produce_plan_infos": [ { "produce_plan_id": produce_plan_info.get("produce_plan_id", ""), "produce_plan_name": produce_plan_info.get("produce_plan_name", ""), "is_success": "绑定成功" if not produce_plan_info.get("msg") else "绑定失败", "msg": produce_plan_info.get("msg", "成功"), } for produce_plan_info in produce_plan_infos ] } }, long_term_memory="Create crawler plan by DouYin Content IDs", ) except Exception as e: logger.error(f"create douyin content crawler plan error. content_ids: {content_ids}, error: {str(e)}") return ToolResult( title="根据抖音内容ID创建爬取计划失败", output="", error=f"创建爬取计划错误:{str(e)}", ) def crawler_plan_bind_produce_plan( input_source_info: Dict[str, Any], produce_plan_ids: List[str], ) -> Tuple[Union[List[Dict[str, str]], None], str]: if not input_source_info or not produce_plan_ids: return None, f"input_source_info or produce_plan_ids is invalid" input_source_check_key = ["inputSourceModal", "inputSourceChannel", "contentType"] try: if not isinstance(produce_plan_ids, list): return None, f"produce_plan_ids is not list" result: List[Dict[str, str]] = [] for produce_plan_id in produce_plan_ids: produce_plan_info = { "produce_plan_id": produce_plan_id, } result.append(produce_plan_info) # 获取生成计划详情，msg不为空表示获取失败 produce_plan_detail_info, msg = find_produce_plan_info_by_id(produce_plan_id) if msg: produce_plan_info["msg"] = msg continue produce_plan_info["produce_plan_name"] = produce_plan_detail_info.get("name", "") input_source_groups = produce_plan_detail_info.get("inputSourceGroups", []) if not input_source_groups: produce_plan_info["msg"] = "生成计划没有输入源组" continue # 查询当前爬取计划要添加到的输入源组下标 input_source_index = 0 for i in range(len(input_source_groups)): input_source_group = input_source_groups[i] if not input_source_group.get("inputSources", []): continue first_input_source = input_source_group.get("inputSources")[0] if all(input_source_info.get(k, 0) == first_input_source.get(k, -1) for k in input_source_check_key): input_source_index = i break # 对应的输入源组添加输入源 input_source_group = input_source_groups[input_source_index] input_source_group.get("inputSources", []).append(input_source_info) response_json = post(PRODUCE_PLAN_SAVE, produce_plan_detail_info) if response_json.get("code") != 0 or not response_json.get("data", {}): produce_plan_info["msg"] = response_json.get("msg", "爬取计划绑定生成计划异常") return result, "" except Exception as e: logger.error(f"crawler_plan_bind_produce_plan error. input_source_info: {json.dumps(input_source_info)}, produce_plan_ids: {produce_plan_ids}, error: {str(e)},") return None, str(e) def find_produce_plan_info_by_id( produce_plan_id: str, ) -> Tuple[Union[Dict[str, str], None], str]: try: if not produce_plan_id or not isinstance(produce_plan_id, str): return None, f"非法的produce_plan_id: {produce_plan_id}" params = { "id": produce_plan_id, } response_json = post(GET_PRODUCE_PLAN_DETAIL_BY_ID, params) if response_json.get("code") != 0 or not response_json.get("data", {}): return None, response_json.get("msg", "获取生成计划详情异常") return response_json.get("data", {}), "" except Exception as e: logger.error(f"find_produce_plan_info_by_id error. produce_plan_id: {produce_plan_id}, error: {str(e)},") return None, str(e) def post(url: str, params: Any) -> Dict[str, Any]: request = { "baseInfo": { "token": DEFAULT_TOKEN, }, "params": params } try: logger.info(f"invoke aigc platform. url: {url}, request: {json.dumps(request)}") response = requests.post( url=url, json=request, headers={"Content-Type": "application/json"}, timeout=DEFAULT_TIMEOUT ) response.raise_for_status() response_json = response.json() logger.info(f"invoke aigc platform. url: {url}, request: {json.dumps(request)}, response: {json.dumps(response_json)}") return response_json except Exception as e: logger.error(f"invoke aigc platform error. url: {url}, request: {json.dumps(request)}, error: {str(e)}") return {}