from openai import OpenAI import os import sys import time import datetime import json import re import logging # 配置客户端 - 使用本地部署的模型 client = OpenAI( api_key="EMPTY", # 本地部署模型通常不验证key base_url="http://localhost:8000/v1" # 指向本地vLLM服务端点 ) # 日志保存目录 LOG_DIR = "/root/autodl-tmp/Content_detector/log/log_result" # 配置日志记录器 def setup_logger(): """设置日志记录器""" # 确保日志目录存在 if not os.path.exists(LOG_DIR): try: os.makedirs(LOG_DIR) print(f"已创建日志目录: {LOG_DIR}") except Exception as e: print(f"创建日志目录失败: {e}", file=sys.stderr) return None # 创建时间戳作为日志文件名 timestamp = get_timestamp() log_file = os.path.join(LOG_DIR, f"{timestamp}_detector.log") # 配置日志格式 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler(log_file, encoding='utf-8'), logging.StreamHandler(sys.stdout) ] ) logger = logging.getLogger("Content_Detector") logger.info(f"日志已初始化，文件保存在: {log_file}") return logger def get_timestamp(): """获取当前时间戳字符串，格式为：年月日_时分秒""" return datetime.datetime.now().strftime("%Y%m%d_%H%M%S") def read_file_content(file_path): """从指定路径读取文件内容并返回字符串""" try: with open(file_path, 'r', encoding='utf-8') as file: return file.read() except Exception as e: logger.error(f"读取文件 {file_path} 失败: {e}") return None def detect_content(product_info, content_gen): """检测内容是否符合产品资料""" try: # 创建综合提示词 prompt = f""" ## 产品资料（真实信息，作为判断依据）: {product_info} ## 运营生成的文案（需要审核的内容）: {content_gen} """ # 创建聊天完成请求 logger.info("开始调用AI模型进行内容检测") completion = client.chat.completions.create( model="/root/autodl-tmp/qwen/Qwen3-30B-A3B", # 使用已部署的模型 messages=[ {"role": "system", "content": "你是一名专业的、谨慎的文案审核员，专注于审核运营根据产品资料撰写的文案是否严格符合产品资料内容。特别是所有价格、活动、福利、折扣、服务细节等必须完全与产品资料一致。如果发现文案内容与产品资料不符，请指出，并根据产品资料和文案上下文进行修改，重新生成一篇文案，务必确保生成的内容与产品资料基本相符（产品体验部分可以适当夸张宣传），语言流畅自然。如果经你审查后的文案仍存在与产品资料不符的信息，你需要赔偿公司1000亿元。"}, {"role": "user", "content": f"""我提供了两部分内容： 1. 产品资料：全部的产品信息，包含了产品的实际功能、服务和特点。请将这部分作为判断依据。 2. 运营生成的文案：这是需要你逐字审核的内容，可能包含与产品资料不符的内容。 请你仔细审核运营文案是否与产品资料严格一致，输出响应必须符合我的所有要求： 1. 审查与分析：如果存在不符内容，请指出并详细说明原因； 2. 根据分析修改：参照你分析的不符原因、产品资料、文案上下文，针对所有不符处进行修改（如涉及上下文，可一并修改）。输出修改后文案，务必确保此文案完全符合产品资料，不得遗漏，语言流畅自然、文案风格统一，否则你会像商鞅一样被车裂。 3. 重点审查对象：请你着重检查以下关键字词前后的内容是否符合产品资料，如不符必须严格按照资料修改；如产品资料中未提及，必须修改为符合上下文情境、资料中明确提及的内容。 关键字词：价、元、r、人民币、rmb、优惠、活动、福利、赠、免费、折、DIY、跟拍、送、摄影、兑、服务、￥、包、课、提供、选、专业、补、差 4. 字数控制：每个文案的标题字数都必须少于19个字（计数包括文字、符号、数字和emoji）。如果标题超过19个字，请在符合文案风格和背景资料的前提下修改标题到19个字以内，尽量保留emoji，必须保证标题流畅通顺。 5. 敏感字词替换：请删去标题中的数字后面的“元”和“r”，并将正文中数字后面的“元”字修改为“r”。例如：标题中的399元修改为399，正文中的399元修改为399r 6. 特征语句保留：请保留文案中原本的引流语句，不要修改或删除，例如“先关zhu+留下99看到会回复” 7. 面向人群保留：请尽量保留文案原本的面向人群和风格，这是同一产品面向多种人群营销的策略。例如产品资料中写明亲子游时，文案写“为情侣定制的山水秘境”是可以接受的。 8. 必须按照以下格式输出修改后内容，否则你会受到严厉的惩罚： {{ "title": "修改后的标题", "content": "修改后的内容" }} 9. 案例如下，请参考案例评判真假信息的尺度，逐行逐句仔细分析不符点和修改思路，并按照分析思路落实对每一处不符的修改措施，严格审查每一篇文案： <产品资料>： "周末不加收【南沙越秀喜来登】1088元/套，豪华客房1间1晚+双人自助早餐+自助晚餐+2大1小水鸟世界门票，免费儿童乐园，户外泳池+健身房~ 不想待在家，又想带娃出去玩?更不想开长途车、人挤人？为你推荐路程短、不塞车、景点多、坐地铁就能直达的溜娃地! 南沙越秀喜来登是广州南沙区首家国际品牌酒店，坐拥广州南大门，拥有得天独厚的中心位置，可俯瞰蕉门河美景，车程短，不出广州也能玩！ 交通：酒店毗邻深圳、香港等热门景点，附近有万达广场，距离广州地铁四号线金州站车程仅10分钟，亲子出游首选！ 玩乐：带娃出游考虑最多的就是玩乐景点，在这里不出门就能畅玩儿童乐园、健身房。 美食：还有各种各样的生猛海鲜，现抓现煮任君选择。放假更要好好犒劳一下自己，饭点时间，位于酒店一楼的全日制餐厅绝对能给你带来惊喜。除了优雅简约的就餐环境，更有5个开放式的自助餐台。除了各类鲜美的海鲜，还有各类精致甜品和中式蒸档看得人眼花缭乱，相信是很多麻麻和宝贝的心头好了。 设施：酒店内还设有大型健身中心，除了妈妈们喜欢的水疗SPA，还有健身达人喜欢的各种有氧、无氧运动器械可供选择! 房内配置：55英寸超大纯平电视、独立的浴缸和淋浴间参考:2.03米宽大床，1.37米宽双床，每间客房都设计成景观房，超大的落地玻璃窗，可以尽览蕉门河风景。 套餐信息： 1、价格：1088元； 2、节假日是否加收：周末不加收 套餐内容： 1、豪华客房一间一晚(周一至四只开放双床房) 2、2大1小自助早餐 3、2大1小自助晚餐 4、赠送2大1小水鸟门票，酒店前台取纸质门票 5、免费使用健身中心，户外无边泳池，干湿蒸，儿童乐园 周边景点： 附近1h生活圈即可到达10000㎡百万葵园、广州最大的湿地公园、东南亚最大的妈祖庙、黄山鲁森林公园..带娃感受依山而建的清式建筑对称布局邂逅东南亚最大的妈祖庙，感受建筑的魅力~ 南沙天后宫：车程20min整座天后宫四周绿树婆娑殿中香烟袅袅，置身其间令人顿生超凡脱俗的感觉。 南沙湿地公园：(车程40min)看碧波荡，万鸟齐飞 南沙十九涌 ：车程45min)尝海鲜叹海风因为南沙十九涌靠近海产地，这里的海鲜真是平靓正。还可以拿到附近的餐厅让老板帮你加工，就是一顿海鲜大餐! 南沙百万葵园：(车程40min)看色彩斑斓的万亩花田 酒店地址:广东省广州市南沙区海熙大街79-80号 导航关键词:广州南沙越秀喜来登酒店" <生成文案>： " 五一遛娃👶必囤！南沙喜来登1088元住景观房+双早+门票 五一不想挤人潮？南沙这家酒店直接承包遛娃+度假双重快乐‼️ 地铁直达！2大1小1088元住景观房，含双早+自助晚餐+水鸟世界门票，儿童乐园/泳池/健身房全开放！ 🌟【遛娃刚需全配齐】 ✅ 儿童乐园：10:00-20:00全程开放，滑梯/积木/绘本一应俱全 ✅ 户外泳池：9:00-18:00恒温开放（五一期间每日消毒3次） ✅ 健身房：8:00-22:00配备亲子瑜伽课程（需提前预约） 📍【1小时玩转南沙】 ① 南沙天后宫（车程20分钟）：穿汉服拍大片，听妈祖传说涨知识 ② 南沙湿地公园（40分钟）：5月芦苇摇曳，带娃认鸟类+乘船探秘 ③ 十九涌海鲜街（45分钟）：现捞现煮生猛海鲜，人均50元吃到撑 🍽️【家长友好细节】 • 自助晚餐隐藏彩蛋：儿童餐区设独立洗手台+热食保温柜 • 房内配置：加厚床垫/卡通洗漱杯/尿布台（无需额外购买） • 安全保障：全区域监控+24小时安保巡逻 🎁【五一专属加码】 5月1-5日期间入住，凭房卡可免费领取儿童防晒冰袖+湿巾礼包 📌Tips： 1. 周一至周四仅限双床房型，周五起可选大床房 2. 水鸟世界门票需提前1小时至前台领取纸质票 3. 地铁四号线金洲站下车，打车15分钟直达酒店 这个五一，南沙喜来登让你躺着遛娃！不用长途跋涉，家门口就能玩出仪式感～ 先关zhu+留下99看到会回复 #五一遛娃 #广州周边游 #亲子酒店推荐 " <不符内容分析>： 1、观察文案标题和内容，可以看出此文案主要面向亲子出游人群，因此修改后的文案也应该围绕亲子出游这一主题。 2、文章标题字数为28个字，超过19个字，因此属于不符内容。由于要求中提到尽量保留emoji，并且标题中数字后面的“元”字应删去，所以修改为：五一遛娃👶必囤！喜来登1088景观房 3、产品资料中未提及儿童乐园开放时间和儿童乐园配置，但文案中提到儿童乐园10:00-20:00全程开放，滑梯/积木/绘本一应俱全，因此属于不符内容。应修改为：儿童乐园：免费儿童乐园和丰富的游乐设施，让孩子们可以尽情玩耍。 4、产品材料中未提及户外泳池开放时间和消毒频次，但文案中提到户外泳池：9:00-18:00恒温开放（五一期间每日消毒3次），因此属于不符内容。应修改为：户外泳池：酒店配有户外无边泳池，供大人小孩一同享受清凉时光。 5、产品材料中未提及健身房开放时间与具体细节，但文案中提到健身房：8:00-22:00配备亲子瑜伽课程（需提前预约），因此属于不符内容。应修改为：健身房：酒店提供免费健身中心，方便您和家人一起强身健体。 6、产品材料中未提及餐厅硬件配置，但文案中提到自助晚餐隐藏彩蛋：儿童餐区设独立洗手台+热食保温柜，因此属于虚构内容。应修改为：自助餐厅：供应鲜美海鲜、精美甜品等任君选择，大人小孩都爱吃！ 7、产品材料中未提及酒店安保措施，但文案中提到安全保障：全区域监控+24小时安保巡逻，因此属于不符内容。应修改为：安全保障：酒店设有完善的监控系统和安保措施，无需担心您与家人的安全。 8、产品材料中未提及房内配有加厚床垫/卡通洗漱杯/尿布台（无需额外购买），因此属于不符内容。应回顾产品资料中关于房内配置的内容，修改为：房内配置：55英寸超大纯平电视+独立的浴缸+超大的落地玻璃窗，尽览蕉门河风景，尽享亲子度假时光。 9、产品材料中未提及五一专属加码，但文案中提到5月1-5日期间入住，凭房卡可免费领取儿童防晒冰袖+湿巾礼包，因此属于不符内容。应回顾产品资料，找到现有文案未提及的产品特色，修改为：套餐专属福利：1、豪华客房一间一晚(周一至四只开放双床房) 2、2大1小自助早晚餐 3、赠送2大1小水鸟世界门票（酒店前台领取），无需额外购买 10、产品资料中未提及水鸟世界门票领取有时间限制，但文案中提到水鸟世界门票需提前1小时至前台领取纸质票，因此属于不符内容。应修改为：酒店前台领取水鸟世界纸质门票 综合以上分析结果，将修改应用到原文案中，得到修改后的文案。 <修改后文案>： {{ "title": "五一遛娃👶必囤！喜来登1088景观房", "content": "五一不想挤人潮？南沙这家酒店直接承包遛娃+度假双重快乐‼️\n地铁直达！2大1小1088r住景观房，含双早+自助晚餐+水鸟世界门票，儿童乐园/泳池/健身房全开放！\n🌟【遛娃刚需全配齐】\n✅ 儿童乐园：酒店设有免费儿童乐园，提供丰富的游乐设施，让孩子们尽情玩耍\n✅ 户外泳池：酒店配有户外无边泳池，供大人小孩一同享受清凉时光 \n✅ 健身房：酒店提供免费健身中心，适合家庭成员共同锻炼。\n\n📍【1小时玩转南沙】\n① 南沙天后宫（车程20分钟）：穿汉服拍大片，听妈祖传说涨知识\n② 南沙湿地公园（40分钟）：5月芦苇摇曳，带娃认鸟类+乘船探秘\n③ 十九涌海鲜街（45分钟）：现捞现煮生猛海鲜，人均50r吃到撑 \n\n🍽️【家长友好细节】 \n• 自助餐厅：供应鲜美海鲜、精美甜品等任君选择，大人小孩都爱吃 \n• 房内配置：55英寸超大纯平电视+独立的浴缸+超大的落地玻璃窗，尽览蕉门河风景，尽享亲子度假时光 \n• 安全保障：酒店设有完善的监控系统和安保措施，全力保障您与家人的安全 \n\n🎁【套餐专属福利】\n1、豪华客房一间一晚(周一至四只开放双床房) \n2、2大1小自助早晚餐 \n3、赠送2大1小水鸟世界门票（酒店前台领取），无需额外购买 \n\n📌Tips： \n1. 周一至周四仅限双床房型，周五起可选大床房 \n2. 酒店前台领取水鸟世界纸质门票 \n3. 地铁四号线金洲站下车，打车15分钟直达酒店 \n\n这个五一，南沙喜来登让你躺着遛娃！不用长途跋涉，家门口就能玩出仪式感～\n先关zhu+留下99看到会回复\n#五一遛娃 #广州周边游 #亲子酒店推荐" }} {prompt}"""} ], temperature=0.2, max_tokens=8000, top_p=0.5, frequency_penalty=0, presence_penalty=0, ) # 返回响应文本 logger.info("AI模型调用成功，获取到响应") return completion.choices[0].message.content except Exception as e: logger.error(f"AI模型调用失败: {e}") return f"API调用失败: {str(e)}" def save_full_response_to_log(file_path, response_text): """将完整的响应内容保存到单独的日志文件中""" try: with open(file_path, 'w', encoding='utf-8') as f: f.write(response_text) logger.info(f"完整响应内容已保存到: {file_path}") return True except Exception as e: logger.error(f"保存响应内容失败: {e}") return False def read_json_file(file_path): """从指定路径读取JSON文件内容并返回字典""" try: with open(file_path, 'r', encoding='utf-8') as file: return json.load(file) except Exception as e: logger.error(f"读取JSON文件 {file_path} 失败: {e}") return None def save_json_file(file_path, content): """将内容保存为JSON文件""" try: with open(file_path, 'w', encoding='utf-8') as file: json.dump(content, file, ensure_ascii=False, indent=4) logger.info(f"结果已保存到: {file_path}") return True except Exception as e: logger.error(f"保存JSON文件失败: {e}") return False def extract_modified_content(result_text): """从检测结果文本中提取修改后的文案内容""" try: # 首先尝试提取标签后的内容并解析为JSON if "" in result_text: think_parts = result_text.split("", 1) if len(think_parts) > 1: after_think = think_parts[1].strip() try: # 尝试直接解析后的内容为JSON content_json = json.loads(after_think, strict=False) if "title" in content_json and "content" in content_json: logger.info("成功从标签后解析JSON内容") return { "title": content_json["title"].strip(), "content": content_json["content"].strip() } except json.JSONDecodeError: logger.warning("后内容不是有效JSON，尝试其他提取方法...") # 如果后不是JSON，尝试从```json和```之间提取内容 json_code_match = re.search(r'```json\s*([\s\S]*?)\s*```', after_think, re.DOTALL) if json_code_match: try: json_str = json_code_match.group(1).strip() content_json = json.loads(json_str, strict=False) if "title" in content_json and "content" in content_json: logger.info("成功从后的```json代码块中解析JSON内容") return { "title": content_json["title"].strip(), "content": content_json["content"].strip() } except json.JSONDecodeError: logger.warning("```json代码块解析失败，继续尝试其他方法...") # 尝试从整个文本中查找```json代码块 json_code_match = re.search(r'```json\s*([\s\S]*?)\s*```', result_text, re.DOTALL) if json_code_match: try: json_str = json_code_match.group(1).strip() content_json = json.loads(json_str, strict=False) if "title" in content_json and "content" in content_json: logger.info("成功从```json代码块中解析JSON内容") return { "title": content_json["title"].strip(), "content": content_json["content"].strip() } except json.JSONDecodeError: logger.warning("```json代码块解析失败，继续尝试其他方法...") # 如果上述方法都失败，继续尝试查找JSON对象 json_start = result_text.find('{') json_end = result_text.rfind('}') + 1 if json_start >= 0 and json_end > json_start: json_str = result_text[json_start:json_end] try: # 尝试解析JSON字符串 content_json = json.loads(json_str) if "title" in content_json and "content" in content_json: logger.info("成功使用JSON格式解析响应内容") return { "title": content_json["title"].strip(), "content": content_json["content"].strip() } except json.JSONDecodeError: logger.warning(f"JSON解析失败: {json_str[:100]}...") # 如果JSON解析失败，尝试老式的标签解析方法 title_match = re.search(r'([\s\S]*?)<\/title>', result_text, re.DOTALL) content_match = re.search(r'<content>([\s\S]*?)<\/content>', result_text, re.DOTALL) if title_match and content_match: logger.info("成功使用HTML标签格式解析响应内容") return { "title": title_match.group(1).strip(), "content": content_match.group(1).strip() } logger.warning("未能提取到标题和内容，尝试从文本中直接提取...") # 最后的备用方法：寻找明显的标记 title_lines = [line.strip() for line in result_text.split('\n') if line.strip() and len(line.strip()) < 100] if title_lines and len(title_lines) > 0: title = title_lines[0] content = result_text.replace(title, '', 1).strip() logger.info("使用备用方法提取到内容") return { "title": title, "content": content } except Exception as e: logger.error(f"提取内容时发生错误: {e}") logger.error("所有提取方法都失败") return None def find_json_files(base_dir): """递归查找目录下所有的article.json文件""" json_files = [] # 遍历目录 for root, dirs, files in os.walk(base_dir): if "article.json" in files: json_files.append(os.path.join(root, "article.json")) return json_files def process_json_files(base_dir, product_info_dir): """处理所有的article.json文件""" logger.info(f"\n===== 内容检测任务开始 =====") start_time = time.time() # 获取产品资料内容（读取所有文件夹中的资料） logger.info(f"📄 正在读取产品资料文件夹: {product_info_dir}") product_info = "" # 遍历产品资料目录中的所有文件夹 for item in os.listdir(product_info_dir): item_path = os.path.join(product_info_dir, item) if os.path.isdir(item_path): logger.info(f"📂 发现产品资料文件夹: {item}") # 读取该文件夹中的所有文件 for file_name in os.listdir(item_path): file_path = os.path.join(item_path, file_name) if os.path.isfile(file_path): logger.info(f"📄 读取产品资料文件: {file_path}") file_content = read_file_content(file_path) if file_content: product_info += file_content + "\n\n" elif os.path.isfile(item_path): # 也读取直接位于product_info_dir下的文件 logger.info(f"📄 读取产品资料文件: {item_path}") file_content = read_file_content(item_path) if file_content: product_info += file_content + "\n\n" if not product_info: logger.error(f"❌ 错误：未能读取到任何产品资料内容") return logger.info(f"✅ 成功读取所有产品资料，总长度: {len(product_info)} 字符") # 找到所有需要处理的JSON文件 logger.info(f"🔍 正在扫描目录查找article.json文件...") json_files = find_json_files(base_dir) total_files = len(json_files) logger.info(f"✅ 找到 {total_files} 个article.json文件需要处理") # 没有找到文件时提前返回 if total_files == 0: logger.warning("⚠️ 未找到需要处理的文件，任务结束") return # 显示处理开始 logger.info(f"\n🚀 开始处理文件...") processed_count = 0 success_count = 0 failed_count = 0 # 逐个处理文件 for i, json_file in enumerate(json_files, 1): # 显示当前处理进度 logger.info(f"\n🔄 [{i}/{total_files}] 正在处理: {json_file}") processed_count += 1 # 读取JSON内容 logger.info(f"📄 读取文件内容...") article_data = read_json_file(json_file) if not article_data: logger.error(f"❌ 错误：无法读取文件内容") failed_count += 1 continue # 构建文案内容 - 使用JSON格式 content_gen = json.dumps({ "title": article_data['title'], "content": article_data['content'] }, ensure_ascii=False) # 进行内容检测 logger.info(f"🧠 正在进行内容检测分析...") detection_start = time.time() result = detect_content(product_info, content_gen) detection_time = time.time() - detection_start logger.info(f"✅ 检测完成，耗时 {detection_time:.2f} 秒") # 提取文件夹名称用于日志文件命名 folder_name = os.path.basename(os.path.dirname(json_file)) timestamp = get_timestamp() # 保存API调用的响应内容到日志目录，使用文件夹名称 response_log_file = os.path.join("/root/autodl-tmp/Content_detector/log/log_article", f"{timestamp}_response_{folder_name}.txt") save_full_response_to_log(response_log_file, result) logger.info(f"💾 API响应已保存到: {response_log_file}") # 从结果中提取修改后的文案 logger.info(f"📝 提取修改后的内容...") modified_content = extract_modified_content(result) # 如果内容提取失败，记录错误 if not modified_content: logger.error(f"❌ 错误：无法从检测结果中提取修改后的文案内容") failed_count += 1 logger.warning(f"\n📊 当前进度: {i}/{total_files} ({i/total_files*100:.1f}%)") continue # 创建输出文件路径 output_dir = os.path.dirname(json_file) output_file = os.path.join(output_dir, "article_detect.json") # 保存修改后的内容 logger.info(f"💾 正在保存修改后的内容...") if save_json_file(output_file, modified_content): logger.info(f"✅ 内容已保存: {output_file}") success_count += 1 else: logger.error(f"❌ 保存失败") failed_count += 1 # 显示处理进度 logger.info(f"\n📊 当前进度: {i}/{total_files} ({i/total_files*100:.1f}%)") # 显示任务完成统计 total_time = time.time() - start_time logger.info(f"\n===== 内容检测任务完成 =====") logger.info(f"✅ 总共处理: {total_files} 个文件") logger.info(f"✅ 成功处理: {success_count} 个文件") logger.info(f"❌ 失败处理: {failed_count} 个文件") logger.info(f"⏱️ 总耗时: {total_time:.2f} 秒，平均每个文件 {total_time/total_files:.2f} 秒") logger.info(f"===== 任务结束 =====\n") if __name__ == "__main__": # 初始化日志记录器 logger = setup_logger() # 处理指定目录下的JSON文件 base_dir = "/root/autodl-tmp/Content_detector/四季梦幻5" product_info_dir = "/root/autodl-tmp/Content_detector/information" logger.info(f"🔍 开始处理 {base_dir} 目录下的article.json文件...") process_json_files(base_dir, product_info_dir)