|
|
@@ -51,6 +51,400 @@ def get_minio_client():
|
|
|
return None
|
|
|
|
|
|
|
|
|
+def process_career_path(career_path, talent_node_id, talent_name_zh):
|
|
|
+ """
|
|
|
+ 处理career_path，创建Hotel节点和相关的Neo4j关系
|
|
|
+
|
|
|
+ Args:
|
|
|
+ career_path (list): 职业轨迹列表，每个元素包含酒店、职位等信息
|
|
|
+ talent_node_id (int): Talent节点的Neo4j ID
|
|
|
+ talent_name_zh (str): 人才中文姓名，用于日志记录
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ dict: 处理结果信息，包含成功和失败的统计
|
|
|
+ """
|
|
|
+ result = {
|
|
|
+ 'total_items': 0,
|
|
|
+ 'hotels_created': 0,
|
|
|
+ 'hotels_skipped': 0,
|
|
|
+ 'brand_relationships_created': 0,
|
|
|
+ 'brand_relationships_failed': 0,
|
|
|
+ 'work_for_relationships_created': 0,
|
|
|
+ 'work_for_relationships_failed': 0,
|
|
|
+ 'work_as_relationships_created': 0,
|
|
|
+ 'work_as_relationships_failed': 0,
|
|
|
+ 'errors': []
|
|
|
+ }
|
|
|
+
|
|
|
+ try:
|
|
|
+ if not career_path or not isinstance(career_path, list):
|
|
|
+ logging.info("career_path为空或不是列表格式，跳过Hotel节点创建")
|
|
|
+ return result
|
|
|
+
|
|
|
+ result['total_items'] = len(career_path)
|
|
|
+ logging.info(f"开始处理career_path，共 {len(career_path)} 条职业轨迹记录")
|
|
|
+
|
|
|
+ # 在执行当前代码逻辑之前，清除传入节点的WORK_FOR和WORK_AS关系
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ # 清除WORK_FOR关系
|
|
|
+ clear_work_for_query = """
|
|
|
+ MATCH (t:Talent)-[r:WORK_FOR]->(h:Hotel)
|
|
|
+ WHERE id(t) = $talent_node_id
|
|
|
+ DELETE r
|
|
|
+ """
|
|
|
+ work_for_result = session.run(clear_work_for_query, talent_node_id=talent_node_id)
|
|
|
+ logging.info(f"已清除Talent节点(ID: {talent_node_id})的所有WORK_FOR关系")
|
|
|
+
|
|
|
+ # 清除WORK_AS关系
|
|
|
+ clear_work_as_query = """
|
|
|
+ MATCH (t:Talent)-[r:WORK_AS]->(d:DataLabel)
|
|
|
+ WHERE id(t) = $talent_node_id
|
|
|
+ DELETE r
|
|
|
+ """
|
|
|
+ work_as_result = session.run(clear_work_as_query, talent_node_id=talent_node_id)
|
|
|
+ logging.info(f"已清除Talent节点(ID: {talent_node_id})的所有WORK_AS关系")
|
|
|
+
|
|
|
+ except Exception as clear_error:
|
|
|
+ logging.error(f"清除Talent节点关系失败: {str(clear_error)}")
|
|
|
+ result['errors'].append(f"清除Talent节点关系失败: {str(clear_error)}")
|
|
|
+ # 即使清除关系失败，也继续执行后续逻辑
|
|
|
+
|
|
|
+ for i, career_item in enumerate(career_path):
|
|
|
+ try:
|
|
|
+ if not isinstance(career_item, dict):
|
|
|
+ logging.warning(f"跳过无效的career_path元素 {i}: 不是字典格式")
|
|
|
+ result['hotels_skipped'] += 1
|
|
|
+ continue
|
|
|
+
|
|
|
+ hotel_zh = career_item.get('hotel_zh', '')
|
|
|
+ hotel_en = career_item.get('hotel_en', '')
|
|
|
+
|
|
|
+ if not hotel_zh:
|
|
|
+ logging.warning(f"跳过career_path元素 {i}: 缺少hotel_zh字段")
|
|
|
+ result['hotels_skipped'] += 1
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 创建Hotel节点
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ # 直接使用Cypher语句查找或创建Hotel节点
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ # 首先查找是否已存在相同hotel_zh的Hotel节点
|
|
|
+ find_query = """
|
|
|
+ MATCH (h:Hotel {hotel_zh: $hotel_zh})
|
|
|
+ RETURN id(h) as node_id, h.hotel_zh as hotel_zh
|
|
|
+ LIMIT 1
|
|
|
+ """
|
|
|
+ find_result = session.run(find_query, hotel_zh=hotel_zh).single()
|
|
|
+
|
|
|
+ if find_result:
|
|
|
+ # 找到现有节点，使用其ID
|
|
|
+ hotel_node_id = find_result['node_id']
|
|
|
+ logging.info(f"找到现有Hotel节点，Neo4j ID: {hotel_node_id}, 酒店: {hotel_zh}")
|
|
|
+ result['hotels_created'] += 0 # 不增加计数，因为不是新创建的
|
|
|
+ else:
|
|
|
+ # 没有找到，创建新节点
|
|
|
+ current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
+ create_query = """
|
|
|
+ CREATE (h:Hotel {
|
|
|
+ hotel_zh: $hotel_zh,
|
|
|
+ hotel_en: $hotel_en,
|
|
|
+ create_time: $create_time
|
|
|
+ })
|
|
|
+ RETURN id(h) as node_id
|
|
|
+ """
|
|
|
+ create_result = session.run(create_query,
|
|
|
+ hotel_zh=hotel_zh,
|
|
|
+ hotel_en=hotel_en,
|
|
|
+ create_time=current_time).single()
|
|
|
+
|
|
|
+ hotel_node_id = create_result['node_id']
|
|
|
+ logging.info(f"成功创建新Hotel节点，Neo4j ID: {hotel_node_id}, 酒店: {hotel_zh}")
|
|
|
+ result['hotels_created'] += 1
|
|
|
+
|
|
|
+ except Exception as hotel_error:
|
|
|
+ logging.error(f"创建Hotel节点失败: {str(hotel_error)}")
|
|
|
+ result['errors'].append(f"创建Hotel节点失败: {hotel_zh}, 错误: {str(hotel_error)}")
|
|
|
+ continue
|
|
|
+
|
|
|
+ # 使用千问大模型判断酒店所属品牌
|
|
|
+ try:
|
|
|
+ from app.core.llm.llm_service import llm_client
|
|
|
+
|
|
|
+ # 构建提示词
|
|
|
+ prompt = f"请根据酒店名称'{hotel_zh}'判断该酒店所属的品牌。请只返回JSON格式结果，格式为{{\"brand\":\"品牌名称\"}}。"
|
|
|
+
|
|
|
+ # 调用千问大模型
|
|
|
+ brand_response = llm_client(prompt)
|
|
|
+
|
|
|
+ if brand_response and isinstance(brand_response, str):
|
|
|
+ # 尝试解析JSON响应
|
|
|
+ try:
|
|
|
+ brand_data = json.loads(brand_response)
|
|
|
+ brand_name = brand_data.get('brand', '')
|
|
|
+
|
|
|
+ if brand_name:
|
|
|
+ # 查找对应的DataLabel节点
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ # 直接查询Neo4j查找name_zh等于品牌名称的DataLabel节点
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ query = "MATCH (n:DataLabel {name_zh: $brand_name}) RETURN id(n) as node_id, n.name_zh as name_zh LIMIT 1"
|
|
|
+ result_query = session.run(query, brand_name=brand_name).single()
|
|
|
+
|
|
|
+ if result_query:
|
|
|
+ label_node_id = result_query['node_id']
|
|
|
+
|
|
|
+ # 在创建BELONGS_TO关系之前，先检查关系是否已经存在
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ # 检查Hotel节点与DataLabel节点之间是否已经存在BELONGS_TO关系
|
|
|
+ check_relationship_query = """
|
|
|
+ MATCH (h:Hotel)-[r:BELONGS_TO]->(d:DataLabel)
|
|
|
+ WHERE id(h) = $hotel_node_id AND id(d) = $label_node_id
|
|
|
+ RETURN r
|
|
|
+ LIMIT 1
|
|
|
+ """
|
|
|
+ existing_relationship = session.run(check_relationship_query,
|
|
|
+ hotel_node_id=hotel_node_id,
|
|
|
+ label_node_id=label_node_id).single()
|
|
|
+
|
|
|
+ if existing_relationship:
|
|
|
+ logging.info(f"Hotel节点与品牌标签的关系已存在，跳过创建: {hotel_zh} BELONGS_TO {brand_name}")
|
|
|
+ result['brand_relationships_created'] += 0 # 不增加计数，因为关系已存在
|
|
|
+ else:
|
|
|
+ # 关系不存在，创建新的BELONGS_TO关系
|
|
|
+ from app.core.graph.graph_operations import create_relationship
|
|
|
+
|
|
|
+ relationship_created = create_relationship(
|
|
|
+ hotel_node_id,
|
|
|
+ label_node_id,
|
|
|
+ 'BELONGS_TO'
|
|
|
+ )
|
|
|
+
|
|
|
+ if relationship_created:
|
|
|
+ logging.info(f"成功创建Hotel节点与品牌标签的关系: {hotel_zh} BELONGS_TO {brand_name}")
|
|
|
+ result['brand_relationships_created'] += 1
|
|
|
+ else:
|
|
|
+ logging.warning(f"创建Hotel节点与品牌标签关系失败: {hotel_zh} -> {brand_name}")
|
|
|
+ result['brand_relationships_failed'] += 1
|
|
|
+
|
|
|
+ except Exception as check_error:
|
|
|
+ logging.error(f"检查Hotel节点与品牌标签关系失败: {str(check_error)}")
|
|
|
+ result['errors'].append(f"检查关系失败: {hotel_zh} -> {brand_name}, 错误: {str(check_error)}")
|
|
|
+ # 即使检查失败，也尝试创建关系
|
|
|
+ from app.core.graph.graph_operations import create_relationship
|
|
|
+
|
|
|
+ relationship_created = create_relationship(
|
|
|
+ hotel_node_id,
|
|
|
+ label_node_id,
|
|
|
+ 'BELONGS_TO'
|
|
|
+ )
|
|
|
+
|
|
|
+ if relationship_created:
|
|
|
+ logging.info(f"成功创建Hotel节点与品牌标签的关系: {hotel_zh} BELONGS_TO {brand_name}")
|
|
|
+ result['brand_relationships_created'] += 1
|
|
|
+ else:
|
|
|
+ logging.warning(f"创建Hotel节点与品牌标签关系失败: {hotel_zh} -> {brand_name}")
|
|
|
+ result['brand_relationships_failed'] += 1
|
|
|
+ else:
|
|
|
+ logging.warning(f"未找到品牌标签节点: {brand_name}")
|
|
|
+
|
|
|
+ except Exception as query_error:
|
|
|
+ logging.error(f"查询品牌标签节点失败: {str(query_error)}")
|
|
|
+ result['errors'].append(f"查询品牌标签节点失败: {brand_name}, 错误: {str(query_error)}")
|
|
|
+ else:
|
|
|
+ logging.warning(f"千问大模型返回的品牌名称为空: {hotel_zh}")
|
|
|
+
|
|
|
+ except json.JSONDecodeError as json_error:
|
|
|
+ logging.warning(f"解析千问大模型返回的JSON失败: {brand_response}, 错误: {json_error}")
|
|
|
+ else:
|
|
|
+ logging.warning(f"千问大模型返回结果无效: {brand_response}")
|
|
|
+
|
|
|
+ except Exception as brand_error:
|
|
|
+ logging.error(f"调用千问大模型判断品牌失败: {str(brand_error)}")
|
|
|
+ result['errors'].append(f"调用千问大模型判断品牌失败: {hotel_zh}, 错误: {str(brand_error)}")
|
|
|
+
|
|
|
+ # 创建Talent节点到Hotel节点的WORK_FOR关系
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import create_relationship
|
|
|
+
|
|
|
+ # 获取职业轨迹信息
|
|
|
+ title_zh = career_item.get('title_zh', '')
|
|
|
+ date = career_item.get('date', '')
|
|
|
+
|
|
|
+ # 创建WORK_FOR关系，包含title_zh和date属性
|
|
|
+ work_for_properties = {}
|
|
|
+ if title_zh:
|
|
|
+ work_for_properties['title_zh'] = title_zh
|
|
|
+ if date:
|
|
|
+ work_for_properties['date'] = date
|
|
|
+
|
|
|
+ # 创建Talent节点到Hotel节点的WORK_FOR关系
|
|
|
+ work_for_relationship_created = create_relationship(
|
|
|
+ talent_node_id, # Talent节点ID
|
|
|
+ hotel_node_id, # Hotel节点ID
|
|
|
+ 'WORK_FOR',
|
|
|
+ work_for_properties
|
|
|
+ )
|
|
|
+
|
|
|
+ if work_for_relationship_created:
|
|
|
+ logging.info(f"成功创建Talent到Hotel的WORK_FOR关系: Talent({talent_name_zh}) WORK_FOR Hotel({hotel_zh})")
|
|
|
+ result['work_for_relationships_created'] += 1
|
|
|
+ else:
|
|
|
+ logging.warning(f"创建Talent到Hotel的WORK_FOR关系失败: Talent({talent_name_zh}) -> Hotel({hotel_zh})")
|
|
|
+ result['work_for_relationships_failed'] += 1
|
|
|
+
|
|
|
+ # 创建Talent节点到DataLabel节点的WORK_AS关系
|
|
|
+ try:
|
|
|
+ # 查找对应的DataLabel节点（职位标签）
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ # 直接查询Neo4j查找name_zh等于title_zh的DataLabel节点
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ query = "MATCH (n:DataLabel {name_zh: $title_zh}) RETURN id(n) as node_id, n.name_zh as name_zh LIMIT 1"
|
|
|
+ result_query = session.run(query, title_zh=title_zh).single()
|
|
|
+
|
|
|
+ if result_query:
|
|
|
+ # 找到现有的DataLabel节点
|
|
|
+ label_node_id = result_query['node_id']
|
|
|
+ logging.info(f"找到现有职位标签节点: {title_zh}, ID: {label_node_id}")
|
|
|
+ else:
|
|
|
+ # 没有找到，创建新的DataLabel节点
|
|
|
+ from app.core.graph.graph_operations import create_or_get_node
|
|
|
+
|
|
|
+ current_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
+ label_properties = {
|
|
|
+ 'name_zh': title_zh,
|
|
|
+ 'en_name': career_item.get('title_en', ''),
|
|
|
+ 'describe': '',
|
|
|
+ 'time': current_time,
|
|
|
+ 'category': '人才地图',
|
|
|
+ 'status': 'active',
|
|
|
+ 'node_type': 'position'
|
|
|
+ }
|
|
|
+
|
|
|
+ label_node_id = create_or_get_node('DataLabel', **label_properties)
|
|
|
+ logging.info(f"创建新职位标签节点: {title_zh}, ID: {label_node_id}")
|
|
|
+
|
|
|
+ # 创建WORK_AS关系，包含hotel_zh和date属性
|
|
|
+ work_as_properties = {}
|
|
|
+ if hotel_zh:
|
|
|
+ work_as_properties['hotel_zh'] = hotel_zh
|
|
|
+ if date:
|
|
|
+ work_as_properties['date'] = date
|
|
|
+
|
|
|
+ # 创建Talent节点到DataLabel节点的WORK_AS关系
|
|
|
+ work_as_relationship_created = create_relationship(
|
|
|
+ talent_node_id, # Talent节点ID
|
|
|
+ label_node_id, # DataLabel节点ID
|
|
|
+ 'WORK_AS',
|
|
|
+ work_as_properties
|
|
|
+ )
|
|
|
+
|
|
|
+ if work_as_relationship_created:
|
|
|
+ logging.info(f"成功创建Talent到职位标签的WORK_AS关系: Talent({talent_name_zh}) WORK_AS DataLabel({title_zh})")
|
|
|
+ result['work_as_relationships_created'] += 1
|
|
|
+ else:
|
|
|
+ logging.warning(f"创建Talent到职位标签的WORK_AS关系失败: Talent({talent_name_zh}) -> DataLabel({title_zh})")
|
|
|
+ result['work_as_relationships_failed'] += 1
|
|
|
+
|
|
|
+ except Exception as label_query_error:
|
|
|
+ logging.error(f"查询或创建职位标签节点失败: {str(label_query_error)}")
|
|
|
+ result['errors'].append(f"查询或创建职位标签节点失败: {title_zh}, 错误: {str(label_query_error)}")
|
|
|
+
|
|
|
+ except Exception as work_as_error:
|
|
|
+ logging.error(f"创建WORK_AS关系失败: {str(work_as_error)}")
|
|
|
+ result['errors'].append(f"创建WORK_AS关系失败: {title_zh}, 错误: {str(work_as_error)}")
|
|
|
+
|
|
|
+ except Exception as work_for_error:
|
|
|
+ logging.error(f"创建WORK_FOR关系失败: {str(work_for_error)}")
|
|
|
+ result['errors'].append(f"创建WORK_FOR关系失败: {hotel_zh}, 错误: {str(work_for_error)}")
|
|
|
+
|
|
|
+ except Exception as career_error:
|
|
|
+ logging.error(f"处理career_path元素 {i} 失败: {str(career_error)}")
|
|
|
+ result['errors'].append(f"处理career_path元素 {i} 失败: {str(career_error)}")
|
|
|
+ continue
|
|
|
+
|
|
|
+ logging.info(f"career_path处理完成，统计信息: {result}")
|
|
|
+ return result
|
|
|
+
|
|
|
+ except Exception as career_path_error:
|
|
|
+ error_msg = f"处理career_path失败: {str(career_path_error)}"
|
|
|
+ logging.error(error_msg)
|
|
|
+ result['errors'].append(error_msg)
|
|
|
+ return result
|
|
|
+
|
|
|
+
|
|
|
+def create_or_get_talent_node(**properties):
|
|
|
+ """
|
|
|
+ 创建具有给定属性的新Talent节点或获取现有节点
|
|
|
+ 如果具有相同pg_id的节点存在，则更新属性
|
|
|
+
|
|
|
+ Args:
|
|
|
+ **properties: 作为关键字参数的节点属性，必须包含pg_id
|
|
|
+
|
|
|
+ Returns:
|
|
|
+ 节点id
|
|
|
+ """
|
|
|
+ try:
|
|
|
+ from app.core.graph.graph_operations import connect_graph
|
|
|
+
|
|
|
+ # 检查是否提供了pg_id
|
|
|
+ if 'pg_id' not in properties:
|
|
|
+ raise ValueError("pg_id is required for Talent node creation")
|
|
|
+
|
|
|
+ pg_id = properties['pg_id']
|
|
|
+
|
|
|
+ with connect_graph().session() as session:
|
|
|
+ # 检查节点是否存在（根据pg_id查找）
|
|
|
+ query = """
|
|
|
+ MATCH (n:Talent {pg_id: $pg_id})
|
|
|
+ RETURN n
|
|
|
+ """
|
|
|
+ result = session.run(query, pg_id=pg_id).single()
|
|
|
+
|
|
|
+ if result:
|
|
|
+ # 节点存在，更新属性
|
|
|
+ props_string = ", ".join([f"n.{key} = ${key}" for key in properties if key != 'pg_id'])
|
|
|
+ if props_string:
|
|
|
+ update_query = f"""
|
|
|
+ MATCH (n:Talent {{pg_id: $pg_id}})
|
|
|
+ SET {props_string}
|
|
|
+ RETURN id(n) as node_id
|
|
|
+ """
|
|
|
+ result = session.run(update_query, **properties).single()
|
|
|
+ logging.info(f"已更新现有Talent节点，pg_id: {pg_id}, Neo4j ID: {result['node_id']}")
|
|
|
+ return result["node_id"]
|
|
|
+ else:
|
|
|
+ # 没有需要更新的属性，返回现有节点ID
|
|
|
+ existing_node_id = result['n'].id
|
|
|
+ logging.info(f"找到现有Talent节点，pg_id: {pg_id}, Neo4j ID: {existing_node_id}")
|
|
|
+ return existing_node_id
|
|
|
+
|
|
|
+ # 如果到这里，则创建新节点
|
|
|
+ props_keys = ", ".join([f"{key}: ${key}" for key in properties])
|
|
|
+ create_query = f"""
|
|
|
+ CREATE (n:Talent {{{props_keys}}})
|
|
|
+ RETURN id(n) as node_id
|
|
|
+ """
|
|
|
+ result = session.run(create_query, **properties).single()
|
|
|
+ logging.info(f"已创建新Talent节点，pg_id: {pg_id}, Neo4j ID: {result['node_id']}")
|
|
|
+ return result["node_id"]
|
|
|
+
|
|
|
+ except Exception as e:
|
|
|
+ logging.error(f"Error in create_or_get_talent_node: {str(e)}")
|
|
|
+ raise e
|
|
|
+
|
|
|
+
|
|
|
def get_parse_tasks(page=1, per_page=10, task_type=None, task_status=None):
|
|
|
"""
|
|
|
获取解析任务列表
|
|
|
@@ -753,9 +1147,6 @@ def add_single_talent(talent_data, minio_path=None, task_type=None):
|
|
|
if new_mobile:
|
|
|
# 如果有新的手机号码，合并到现有手机号码中
|
|
|
existing_card.mobile = merge_mobile_numbers(existing_card.mobile, new_mobile)
|
|
|
- elif talent_data.get('mobile') == '':
|
|
|
- # 如果明确传入空字符串，则清空手机号码
|
|
|
- existing_card.mobile = ''
|
|
|
|
|
|
existing_card.phone = talent_data.get('phone', existing_card.phone)
|
|
|
existing_card.email = talent_data.get('email', existing_card.email)
|
|
|
@@ -812,22 +1203,45 @@ def add_single_talent(talent_data, minio_path=None, task_type=None):
|
|
|
|
|
|
# 在Neo4j图数据库中更新Talent节点
|
|
|
try:
|
|
|
- from app.core.graph.graph_operations import create_or_get_node
|
|
|
-
|
|
|
# 创建Talent节点属性
|
|
|
talent_properties = {
|
|
|
'name_zh': existing_card.name_zh,
|
|
|
'name_en': existing_card.name_en,
|
|
|
'mobile': existing_card.mobile,
|
|
|
+ 'phone': existing_card.phone,
|
|
|
'email': existing_card.email,
|
|
|
+ 'status': existing_card.status,
|
|
|
+ 'birthday': existing_card.birthday.strftime('%Y-%m-%d') if existing_card.birthday else None,
|
|
|
+ 'age': existing_card.age,
|
|
|
+ 'residence': existing_card.residence,
|
|
|
+ 'native_place': existing_card.native_place,
|
|
|
'pg_id': existing_card.id, # PostgreSQL主记录的ID
|
|
|
'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
}
|
|
|
|
|
|
# 在Neo4j中更新或创建Talent节点
|
|
|
- neo4j_node_id = create_or_get_node('Talent', **talent_properties)
|
|
|
+ neo4j_node_id = create_or_get_talent_node(**talent_properties)
|
|
|
logging.info(f"成功在Neo4j中更新Talent节点，Neo4j ID: {neo4j_node_id}, PostgreSQL ID: {existing_card.id}")
|
|
|
|
|
|
+ # 处理career_path，创建相关的Neo4j节点和关系
|
|
|
+ if existing_card.career_path and isinstance(existing_card.career_path, list):
|
|
|
+ try:
|
|
|
+ # 调用process_career_path函数处理职业轨迹
|
|
|
+ career_result = process_career_path(
|
|
|
+ career_path=existing_card.career_path,
|
|
|
+ talent_node_id=neo4j_node_id,
|
|
|
+ talent_name_zh=existing_card.name_zh
|
|
|
+ )
|
|
|
+
|
|
|
+ # 记录处理结果
|
|
|
+ logging.info(f"处理career_path完成，结果: {career_result}")
|
|
|
+
|
|
|
+ except Exception as career_error:
|
|
|
+ logging.error(f"处理career_path失败: {str(career_error)}")
|
|
|
+ # career_path处理失败不影响主流程，继续执行
|
|
|
+ else:
|
|
|
+ logging.info(f"人才记录 {existing_card.id} 没有career_path数据，跳过Neo4j关系处理")
|
|
|
+
|
|
|
# 更新parsed_talents表中的对应记录状态
|
|
|
if talent_data.get('id'):
|
|
|
try:
|
|
|
@@ -866,40 +1280,22 @@ def add_single_talent(talent_data, minio_path=None, task_type=None):
|
|
|
main_card.origin_source = _update_origin_source_with_minio_path(main_card.origin_source, talent_data)
|
|
|
db.session.commit() # 提交origin_source的更新
|
|
|
|
|
|
- # 在Neo4j图数据库中创建Talent节点
|
|
|
- try:
|
|
|
- from app.core.graph.graph_operations import create_or_get_node
|
|
|
-
|
|
|
- # 创建Talent节点属性
|
|
|
- talent_properties = {
|
|
|
- 'name_zh': main_card.name_zh,
|
|
|
- 'name_en': main_card.name_en,
|
|
|
- 'mobile': main_card.mobile,
|
|
|
- 'email': main_card.email,
|
|
|
- 'pg_id': main_card.id, # PostgreSQL主记录的ID
|
|
|
- 'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
- }
|
|
|
-
|
|
|
- # 在Neo4j中创建Talent节点
|
|
|
- neo4j_node_id = create_or_get_node('Talent', **talent_properties)
|
|
|
- logging.info(f"成功在Neo4j中创建Talent节点，Neo4j ID: {neo4j_node_id}, PostgreSQL ID: {main_card.id}")
|
|
|
-
|
|
|
- # 更新parsed_talents表中的对应记录状态
|
|
|
- if talent_data.get('id'):
|
|
|
- try:
|
|
|
- from app.core.data_parse.parse_system import ParsedTalent
|
|
|
- parsed_talent = ParsedTalent.query.filter_by(id=talent_data['id']).first()
|
|
|
- if parsed_talent:
|
|
|
- parsed_talent.status = '已入库'
|
|
|
- db.session.commit()
|
|
|
- logging.info(f"已更新parsed_talents表记录状态为已入库，ID: {talent_data['id']}")
|
|
|
- except Exception as update_error:
|
|
|
- logging.error(f"更新parsed_talents表记录状态失败: {str(update_error)}")
|
|
|
- # 状态更新失败不影响主流程
|
|
|
-
|
|
|
- except Exception as neo4j_error:
|
|
|
- logging.error(f"在Neo4j中创建Talent节点失败: {str(neo4j_error)}")
|
|
|
- # Neo4j操作失败不影响主流程，继续返回成功结果
|
|
|
+ # 注意：当创建新记录作为主记录并保存疑似重复记录信息时，不在Neo4j图数据库中创建Talent节点
|
|
|
+ # 这是因为疑似重复记录需要进一步人工确认和处理
|
|
|
+ logging.info(f"跳过Neo4j Talent节点创建，等待疑似重复记录处理完成，PostgreSQL ID: {main_card.id}")
|
|
|
+
|
|
|
+ # 更新parsed_talents表中的对应记录状态
|
|
|
+ if talent_data.get('id'):
|
|
|
+ try:
|
|
|
+ from app.core.data_parse.parse_system import ParsedTalent
|
|
|
+ parsed_talent = ParsedTalent.query.filter_by(id=talent_data['id']).first()
|
|
|
+ if parsed_talent:
|
|
|
+ parsed_talent.status = '已入库'
|
|
|
+ db.session.commit()
|
|
|
+ logging.info(f"已更新parsed_talents表记录状态为已入库，ID: {talent_data['id']}")
|
|
|
+ except Exception as update_error:
|
|
|
+ logging.error(f"更新parsed_talents表记录状态失败: {str(update_error)}")
|
|
|
+ # 状态更新失败不影响主流程
|
|
|
|
|
|
return {
|
|
|
'code': 202, # Accepted，表示已接受但需要进一步处理
|
|
|
@@ -966,22 +1362,30 @@ def add_single_talent(talent_data, minio_path=None, task_type=None):
|
|
|
|
|
|
# 在Neo4j图数据库中创建Talent节点
|
|
|
try:
|
|
|
- from app.core.graph.graph_operations import create_or_get_node
|
|
|
-
|
|
|
# 创建Talent节点属性
|
|
|
talent_properties = {
|
|
|
'name_zh': business_card.name_zh,
|
|
|
'name_en': business_card.name_en,
|
|
|
'mobile': business_card.mobile,
|
|
|
+ 'phone': business_card.phone,
|
|
|
'email': business_card.email,
|
|
|
+ 'status': business_card.status,
|
|
|
+ 'birthday': business_card.birthday.strftime('%Y-%m-%d') if business_card.birthday else None,
|
|
|
+ 'age': business_card.age,
|
|
|
+ 'residence': business_card.residence,
|
|
|
+ 'native_place': business_card.native_place,
|
|
|
'pg_id': business_card.id, # PostgreSQL主记录的ID
|
|
|
'updated_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S')
|
|
|
}
|
|
|
|
|
|
# 在Neo4j中创建Talent节点
|
|
|
- neo4j_node_id = create_or_get_node('Talent', **talent_properties)
|
|
|
+ neo4j_node_id = create_or_get_talent_node(**talent_properties)
|
|
|
logging.info(f"成功在Neo4j中创建Talent节点，Neo4j ID: {neo4j_node_id}, PostgreSQL ID: {business_card.id}")
|
|
|
|
|
|
+ # 处理career_path，创建Hotel节点及关系
|
|
|
+ career_result = process_career_path(career_path, neo4j_node_id, business_card.name_zh)
|
|
|
+ logging.info(f"career_path处理完成，结果: {career_result}")
|
|
|
+
|
|
|
# 更新parsed_talents表中的对应记录状态
|
|
|
if talent_data.get('id'):
|
|
|
try:
|
maxiaolong