""" 插入模拟面试数据到 Coze 数据库 通过工作流 C1 执行 SQL INSERT 语句 """ import asyncio import httpx import json import random from datetime import datetime, timedelta # Coze 配置 COZE_API_BASE = "https://api.coze.cn" COZE_PAT_TOKEN = "pat_nd1wU47WyPS9GCIyJ1clnH8h1WOQXGrYELX8w73TnSZaYbFdYD4swIhzcETBUbfT" WORKFLOW_ID = "7597376294612107318" # 查询工作流 ID # 模拟候选人数据 MOCK_CANDIDATES = [ {"name": "张小美", "score_base": 85}, {"name": "李明辉", "score_base": 78}, {"name": "王晓丽", "score_base": 92}, {"name": "陈建国", "score_base": 65}, {"name": "刘芳芳", "score_base": 72}, {"name": "赵大伟", "score_base": 58}, {"name": "孙婷婷", "score_base": 88}, {"name": "周志强", "score_base": 45}, ] # 模拟简历内容 MOCK_RESUMES = [ "3年销售经验，曾在某知名医美机构担任销售顾问，业绩连续12个月达成率超过120%。熟悉轻医美产品线，擅长客户关系维护。", "5年美容行业从业经验，持有高级美容师证书。性格开朗，沟通能力强，善于发现客户需求并提供专业建议。", "应届毕业生，市场营销专业，在校期间有丰富的社团活动经验。对医美行业充满热情，学习能力强。", "2年电商销售经验，熟悉线上运营和客户服务。希望转型到线下医美销售领域，有较强的目标感和执行力。", ] # 模拟评估报告模板 SKILL_REPORTS = [ "候选人展现出扎实的销售基础，在产品介绍和需求挖掘方面表现突出。能够灵活运用 FAB 法则，善于建立客户信任。建议加强异议处理技巧的训练。得分：{score}分", "销售技能较为全面，尤其在客户沟通和关系建立方面有独到之处。能够准确把握客户心理，提供针对性的解决方案。需要在成交技巧上进一步打磨。得分：{score}分", "基础销售技能掌握良好，但缺乏实战经验。理论知识扎实，需要更多的实践机会来提升。建议安排老带新的培训模式。得分：{score}分", ] CONCEPT_REPORTS = [ "对销售工作有正确的认知，理解以客户为中心的服务理念。能够平衡业绩压力和客户满意度，具有长期发展的潜力。得分：{score}分", "销售观念较为成熟，认同医美行业的价值观。注重专业性和诚信度，能够为客户提供真诚的建议。得分：{score}分", "对销售工作的理解还停留在表面，需要加强对行业和产品的深入学习。建议多参与案例分析和行业培训。得分：{score}分", ] COMPETENCY_REPORTS = [ "综合素质优秀，具备良好的学习能力和抗压能力。工作态度积极，团队协作意识强。是值得培养的优秀人才。得分：{score}分", "整体素质良好，有较强的责任心和执行力。沟通表达清晰，形象气质佳。建议加强时间管理能力。得分：{score}分", "基本素质合格，但在主动性和自驱力方面有提升空间。需要更多的激励和引导来激发潜力。得分：{score}分", ] MOTIVATION_SUMMARIES = [ "候选人对轻医美行业表现出浓厚的兴趣，希望在这个领域长期发展。主要动机包括：1）看好行业发展前景；2）认同公司品牌价值；3）期待获得专业成长机会。", "求职动机明确，主要是希望获得更好的收入和职业发展平台。对销售工作有热情，愿意接受挑战和压力。", "动机较为单一，主要关注薪资待遇。建议在后续面试中进一步了解其对行业的认知和长期规划。", ] RISK_WARNINGS = [ "", # 无风险 "", # 无风险 "候选人在上一份工作中存在频繁跳槽的情况，需要关注稳定性问题。建议在背调中重点核实离职原因。", "面试过程中发现候选人对公司产品了解不够深入，可能存在准备不充分的情况。建议安排二次面试进一步评估。", "候选人期望薪资与市场水平存在一定差距，需要在 Offer 阶段进行合理沟通。", ] GROWTH_PLANS = [ "建议培养方向：1）前3个月重点学习产品知识和销售流程；2）3-6个月参与实战，由资深顾问带教；3）6个月后独立负责客户。预计1年内可成长为骨干销售。", "该候选人具备快速成长的潜力，建议：1）入职即安排系统培训；2）重点培养其客户开发能力；3）可考虑作为储备管理人才培养。", "成长空间有限，建议先观察3个月试用期表现再做进一步评估。", ] REF_CHECK_LISTS = [ "建议背调问题：\n1. 请确认候选人在贵公司的任职时间和职位\n2. 候选人的主要工作职责是什么？\n3. 您如何评价候选人的销售能力？\n4. 候选人的离职原因是什么？\n5. 如果有机会，您是否愿意再次与其共事？", "重点背调事项：\n1. 核实业绩数据的真实性\n2. 了解团队协作情况\n3. 确认是否有劳动纠纷\n4. 验证学历和证书的真实性", ] # 模拟对话记录 MOCK_DIALOGUES = [ { "stage": "开场", "ai_question": "您好！欢迎参加我们的 AI 面试。我是您的 AI 面试官，接下来我们将进行一次关于销售岗位的面试。首先，请您简单介绍一下自己。", "user_answer": "您好，我叫{name}，今年28岁。我有3年的销售工作经验，主要在美容行业从事客户服务和销售工作。我性格开朗，善于与人沟通，对轻医美行业非常感兴趣。" }, { "stage": "销售技能", "ai_question": "很好，感谢您的自我介绍。接下来我想了解一下您的销售经验。请您描述一次成功的销售案例，包括您是如何发现客户需求、如何推荐产品、以及最终如何促成交易的。", "user_answer": "好的。去年我接待了一位40多岁的女士，她最初只是来咨询皮肤保养。通过聊天我发现她其实对法令纹比较在意。我没有直接推销，而是先帮她做了皮肤检测，用数据说明问题。然后根据她的预算和需求，推荐了适合的玻尿酸填充方案。整个过程我注重建立信任，最终她不仅做了法令纹填充，还办了年卡。" }, { "stage": "销售技能", "ai_question": "非常好的案例分享。那么当客户对价格有异议时，您通常如何处理？", "user_answer": "价格异议是很常见的。我通常会先认同客户的感受，然后分析价值而不是价格。比如我会说'您的担心我理解，选择医美确实需要慎重考虑。不过您看，我们使用的是进口正品，由资深医生操作，术后还有专业跟踪服务。很多客户反馈效果能保持1-2年，算下来其实性价比很高。'另外我也会提供分期付款等灵活方案。" }, { "stage": "销售观", "ai_question": "您如何看待销售工作？您认为一个优秀的销售顾问最重要的品质是什么？", "user_answer": "我认为销售的本质是帮助客户解决问题，而不是单纯的推销产品。一个优秀的销售顾问首先要专业，要真正了解产品和客户需求；其次要真诚，不能为了业绩误导客户；最后要有服务意识，把每一位客户都当作长期朋友来维护。" }, { "stage": "素质项", "ai_question": "假设您在工作中遇到了连续三个月业绩不达标的情况，您会如何应对？", "user_answer": "首先我会分析原因，是市场问题、个人方法问题还是其他因素。然后我会主动向业绩好的同事请教，学习他们的成功经验。同时我会调整自己的工作方法，比如增加客户回访频率、优化话术等。最重要的是保持积极心态，相信通过努力一定能改善。" }, { "stage": "求职动机", "ai_question": "最后一个问题，您为什么选择我们公司？您对未来的职业发展有什么规划？", "user_answer": "选择贵公司主要有三个原因：一是贵公司在轻医美领域有很好的口碑和品牌影响力；二是听说公司非常注重员工培训和成长；三是公司的企业文化和我的价值观很匹配。未来我希望能在1-2年内成长为资深销售顾问，3年内有机会带领小团队，为公司创造更大价值。" }, ] async def execute_sql(sql: str, table: str) -> dict: """通过工作流执行 SQL""" headers = { "Authorization": f"Bearer {COZE_PAT_TOKEN}", "Content-Type": "application/json" } payload = { "workflow_id": WORKFLOW_ID, "parameters": { "input": json.dumps({"table": table, "sql": sql}, ensure_ascii=False) } } async with httpx.AsyncClient(timeout=60.0) as client: response = await client.post( f"{COZE_API_BASE}/v1/workflow/run", headers=headers, json=payload ) return response.json() def generate_session_id(name: str, idx: int) -> str: """生成 session_id""" timestamp = datetime.now().strftime("%Y%m%d%H%M") return f"SESS_{timestamp}_{name}_{idx:02d}" def generate_assessment_sql(session_id: str, candidate: dict, idx: int) -> str: """生成评估记录 INSERT SQL""" base = candidate["score_base"] variation = random.randint(-5, 10) skill_score = min(100, max(0, base + random.randint(-8, 8))) concept_score = min(100, max(0, base + random.randint(-5, 10))) competency_score = min(100, max(0, base + random.randint(-10, 5))) skill_report = random.choice(SKILL_REPORTS).format(score=skill_score) concept_report = random.choice(CONCEPT_REPORTS).format(score=concept_score) competency_report = random.choice(COMPETENCY_REPORTS).format(score=competency_score) motivation = random.choice(MOTIVATION_SUMMARIES) risk = random.choice(RISK_WARNINGS) growth = random.choice(GROWTH_PLANS) ref_check = random.choice(REF_CHECK_LISTS) resume = random.choice(MOCK_RESUMES) # 转义单引号 def escape(s): return s.replace("'", "''") if s else "" sql = f"""INSERT INTO ci_interview_assessments ( session_id, candidate_name, current_stage, sales_skill_score, sales_skill_report, sales_concept_score, sales_concept_report, competency_score, competency_report, motivation_summary, risk_warning, growth_plan, ref_check_list, resume_text ) VALUES ( '{session_id}', '{candidate["name"]}', '10', '{skill_score}', '{escape(skill_report)}', '{concept_score}', '{escape(concept_report)}', '{competency_score}', '{escape(competency_report)}', '{escape(motivation)}', '{escape(risk)}', '{escape(growth)}', '{escape(ref_check)}', '{escape(resume)}' )""" return sql def generate_log_sql(session_id: str, candidate_name: str, round_num: int, dialogue: dict, idx: int) -> str: """生成对话记录 INSERT SQL""" import uuid def escape(s): return s.replace("'", "''").replace("{name}", candidate_name) if s else "" # 生成唯一的 log_id log_id = f"LOG_{idx:02d}_{round_num:02d}_{uuid.uuid4().hex[:8]}" sql = f"""INSERT INTO ci_interview_logs ( log_id, session_id, stage, round, ai_question, user_answer, log_type ) VALUES ( '{log_id}', '{session_id}', '{dialogue["stage"]}', '{round_num}', '{escape(dialogue["ai_question"])}', '{escape(dialogue["user_answer"])}', 'interview' )""" return sql async def main(): print("=" * 60) print("开始插入模拟面试数据...") print("=" * 60) success_count = 0 error_count = 0 for idx, candidate in enumerate(MOCK_CANDIDATES): session_id = generate_session_id(candidate["name"], idx) print(f"\n[{idx + 1}/{len(MOCK_CANDIDATES)}] 处理候选人: {candidate['name']}") print(f" Session ID: {session_id}") # 1. 插入评估记录 try: assessment_sql = generate_assessment_sql(session_id, candidate, idx) print(f" 插入评估记录...") result = await execute_sql(assessment_sql, "assessments") if result.get("code") == 0: print(f" ✓ 评估记录插入成功") success_count += 1 else: print(f" ✗ 评估记录插入失败: {result}") error_count += 1 except Exception as e: print(f" ✗ 评估记录插入异常: {e}") error_count += 1 # 2. 插入对话记录 for round_num, dialogue in enumerate(MOCK_DIALOGUES, 1): try: log_sql = generate_log_sql(session_id, candidate["name"], round_num, dialogue, idx) print(f" 插入对话记录 (第{round_num}轮)...") result = await execute_sql(log_sql, "logs") if result.get("code") == 0: print(f" ✓ 对话记录 {round_num} 插入成功") success_count += 1 else: print(f" ✗ 对话记录 {round_num} 插入失败: {result}") error_count += 1 except Exception as e: print(f" ✗ 对话记录 {round_num} 插入异常: {e}") error_count += 1 # 等待一下避免请求过快 await asyncio.sleep(1) print("\n" + "=" * 60) print(f"数据插入完成！") print(f"成功: {success_count} 条") print(f"失败: {error_count} 条") print("=" * 60) if __name__ == "__main__": asyncio.run(main())