import json import django import os import sys import pandas as pd import requests from django.db import transaction from django.db.models import Q from django.db import connection from datetime import datetime import traceback import tempfile import uuid # 设置 Django 环境 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'role_based_system.settings') django.setup() import lark_oapi as lark from lark_oapi.api.bitable.v1 import * from user_management.models import FeishuCreator, Data, KnowledgeBase, KnowledgeBaseDocument from django.conf import settings import logging from django.contrib.auth import get_user_model logger = logging.getLogger(__name__) # SDK 使用说明: https://open.feishu.cn/document/uAjLw4CM/ukTMukTMukTM/server-side-sdk/python--sdk/preparations-before-development # 以下示例代码默认根据文档示例值填充,如果存在代码问题,请在 API 调试台填上相关必要参数后再复制代码使用 def extract_field_value(field_value): """提取字段值""" if isinstance(field_value, list): if field_value and isinstance(field_value[0], dict): return field_value[0].get('text', '') elif isinstance(field_value, dict): if 'text' in field_value: return field_value['text'] elif 'link' in field_value: return field_value['link'] elif 'link_record_ids' in field_value: return '' return field_value def save_to_database(record): """从飞书多维表格保存记录到数据库""" fields = record.fields record_id = record.record_id creator_data = { 'record_id': record_id, 'contact_person': extract_field_value(fields.get('对接人', '')), 'handle': extract_field_value(fields.get('Handle', '')), 'tiktok_url': extract_field_value(fields.get('链接', '')), 'fans_count': extract_field_value(fields.get('粉丝数', '')), 'gmv': fields.get('GMV', ''), 'email': extract_field_value(fields.get('邮箱', '')), 'phone': extract_field_value(fields.get('手机号|WhatsApp', '')), 'account_type': extract_field_value(fields.get('账号属性', [])), 'price_quote': fields.get('报价', ''), 'response_speed': fields.get('回复速度', ''), 'cooperation_intention': fields.get('合作意向', ''), 'payment_method': fields.get('支付方式', ''), 'payment_account': fields.get('收款账号', ''), 'address': fields.get('收件地址', ''), 'has_ooin': fields.get('签约OOIN?', ''), 'source': fields.get('渠道来源', ''), 'contact_status': fields.get('建联进度', ''), 'cooperation_brands': fields.get('合作品牌', []), 'system_categories': extract_field_value(fields.get('系统展示的带货品类', [])), 'actual_categories': extract_field_value(fields.get('实际高播放量带货品类', [])), 'human_categories': fields.get('达人标想要货品类', ''), 'creator_base': '', 'notes': extract_field_value(fields.get('父记录', '')), } try: creator, created = FeishuCreator.objects.update_or_create( record_id=record_id, defaults=creator_data ) return creator, created except Exception as e: print(f"保存记录时出错: {str(e)}") print(f"记录数据: {creator_data}") return None, False def fetch_all_records(client, app_token, table_id, user_access_token): """获取所有记录""" total_records = [] page_token = None page_size = 20 while True: try: # 构造请求对象 builder = SearchAppTableRecordRequest.builder() \ .app_token(app_token) \ .table_id(table_id) \ .page_size(page_size) # 如果有page_token,添加到请求中 if page_token: builder = builder.page_token(page_token) # 构建完整请求 request = builder.request_body(SearchAppTableRecordRequestBody.builder().build()).build() print(f"发送请求,page_token: {page_token}") # 发起请求 option = lark.RequestOption.builder().user_access_token(user_access_token).build() response = client.bitable.v1.app_table_record.search(request, option) if not response.success(): print(f"请求失败: {response.code}, {response.msg}") break # 获取当前页记录 current_records = response.data.items if not current_records: print("没有更多记录") break total_records.extend(current_records) # 解析响应数据获取分页信息 response_data = json.loads(response.raw.content) total = response_data["data"]["total"] print(f"获取到 {len(current_records)} 条记录,当前总计: {len(total_records)}/{total} 条") # 获取下一页token page_token = response_data["data"].get("page_token") if not page_token or not response_data["data"].get("has_more", False): print("已获取所有数据") break except Exception as e: print(f"错误: {str(e)}") import traceback print(traceback.format_exc()) break print(f"最终获取到 {len(total_records)} 条记录") return total_records def update_from_excel(excel_file_path): """从Excel文件更新数据库记录""" try: print(f"开始读取Excel文件: {excel_file_path}") df = pd.read_excel(excel_file_path) if 'Handle' not in df.columns: print("错误: Excel文件中缺少'Handle'列") return update_count = 0 skip_count = 0 error_count = 0 # 获取可更新的字段列表 excluded_fields = {'id', 'record_id', 'created_at', 'updated_at'} model_fields = {f.name for f in FeishuCreator._meta.get_fields()} - excluded_fields valid_columns = set(df.columns) & model_fields print(f"可更新的列: {valid_columns}") with transaction.atomic(): for index, row in df.iterrows(): try: handle = str(row['Handle']).strip() if not handle: print(f"跳过第{index+2}行: Handle为空") skip_count += 1 continue # 查找现有记录 creator = FeishuCreator.objects.filter(handle=handle).first() if not creator: print(f"跳过Handle为'{handle}'的记录: 数据库中不存在") skip_count += 1 continue # 准备更新数据 update_data = {} for column in valid_columns: if column == 'Handle': continue value = row[column] if pd.isna(value): continue # 处理特殊类型 if isinstance(value, (list, dict)): value = json.dumps(value) elif isinstance(value, (int, float)): if column in ['fans_count']: value = int(value) else: value = str(value) else: value = str(value).strip() if value: update_data[column] = value # 更新记录 if update_data: for field, value in update_data.items(): setattr(creator, field, value) creator.save() update_count += 1 print(f"已更新Handle为'{handle}'的记录") else: skip_count += 1 print(f"跳过Handle为'{handle}'的记录: 无需更新") except Exception as e: error_count += 1 print(f"处理Handle'{handle}'时出错: {str(e)}") print("\nExcel更新完成!统计信息:") print(f"更新记录数:{update_count}") print(f"跳过记录数:{skip_count}") print(f"错误记录数:{error_count}") except Exception as e: print(f"处理Excel文件时出错: {str(e)}") def sync_from_feishu(app_token=None, table_id=None, user_access_token=None, filters=None): """从飞书同步数据到系统 参数: app_token: 飞书应用的APP_TOKEN,如不提供则使用默认值 table_id: 飞书多维表格的TABLE_ID,如不提供则使用默认值 user_access_token: 用户访问令牌,如不提供则使用默认值 filters: 筛选条件,字典格式 返回: dict: 包含同步统计信息的字典 """ try: # 使用传入的参数或默认值 app_token = app_token or settings.FEISHU_APP_TOKEN table_id = table_id or settings.FEISHU_TABLE_ID user_access_token = user_access_token or settings.FEISHU_USER_ACCESS_TOKEN if not all([app_token, table_id, user_access_token]): logger.error("飞书API凭据不完整") return { 'created': 0, 'updated': 0, 'errors': 1, 'total': 0, 'error_message': '飞书API凭据不完整' } # 初始化统计信息 created_count = 0 updated_count = 0 error_count = 0 created_creators = [] # 获取飞书数据 client = lark.Client.builder() \ .enable_set_token(True) \ .log_level(lark.LogLevel.DEBUG) \ .build() print("开始从飞书同步数据...") all_records = fetch_all_records(client, app_token, table_id, user_access_token) if not all_records: print("未获取到任何记录") return { 'created': 0, 'updated': 0, 'errors': 0, 'total': 0 } print("\n开始更新数据库...") for record in all_records: creator, created = save_to_database(record) if creator: if created: created_count += 1 if created_count % 10 == 0: print(f"已创建 {created_count} 条记录...") created_creators.append(creator) else: updated_count += 1 if updated_count % 10 == 0: print(f"已更新 {updated_count} 条记录...") else: error_count += 1 print(f"处理记录失败") print("\n飞书同步完成!统计信息:") print(f"新建记录:{created_count}") print(f"更新记录:{updated_count}") print(f"错误记录:{error_count}") print(f"总记录数:{len(all_records)}") # 返回统计信息,增加了created_creators列表 return { 'created': created_count, 'updated': updated_count, 'errors': error_count, 'total': created_count + updated_count, 'created_creators': created_creators # 新增:返回新创建的记录对象列表 } except Exception as e: logger.error(f"从飞书同步数据时出错: {str(e)}") logger.error(traceback.format_exc()) return { 'created': 0, 'updated': 0, 'errors': 1, 'total': 0, 'error_message': str(e) } def format_field_value(value, field_type='text'): """格式化字段值以符合飞书多维表格API要求""" if value is None or value == '': return None if field_type == 'text': return {"text": str(value)} elif field_type == 'number': try: return float(value) except (ValueError, TypeError): return None elif field_type == 'checkbox': return value.lower() in ('true', 'yes', '是', '1') elif field_type == 'multi_select': # 多选项需要返回一个选项ID列表 if isinstance(value, str): items = [item.strip() for item in value.split(',') if item.strip()] return [{"text": item} for item in items] elif isinstance(value, list): return [{"text": str(item)} for item in value] return [] return {"text": str(value)} def export_to_feishu(query_set=None): """从数据库导出记录到飞书多维表格""" # 创建client client = lark.Client.builder() \ .enable_set_token(True) \ .log_level(lark.LogLevel.DEBUG) \ .build() # 配置参数 APP_TOKEN = "XYE6bMQUOaZ5y5svj4vcWohGnmg" TABLE_ID = "tbl3oikG3F8YYtVA" USER_ACCESS_TOKEN = "u-ecM5BmzKx4uHz3sG0FouQSk1l9kxgl_3Xa00l5Ma24Jy" print("开始从数据库导出数据到飞书...") # 如果没有提供查询集,则使用所有没有record_id的记录 if not query_set: query_set = FeishuCreator.objects.filter(record_id='') if not query_set.exists(): print("没有需要导出的记录") return created_count = 0 updated_count = 0 error_count = 0 for creator in query_set: try: # 准备字段数据 fields = { "对接人": format_field_value(creator.contact_person), "Handle": format_field_value(creator.handle), "链接": format_field_value(creator.tiktok_url), "粉丝数": format_field_value(creator.fans_count), "GMV": format_field_value(creator.gmv), "邮箱": format_field_value(creator.email), "手机号|WhatsApp": format_field_value(creator.phone), "账号属性": format_field_value(creator.account_type, 'multi_select'), "报价": format_field_value(creator.price_quote), "回复速度": format_field_value(creator.response_speed), "合作意向": format_field_value(creator.cooperation_intention), "支付方式": format_field_value(creator.payment_method), "收款账号": format_field_value(creator.payment_account), "收件地址": format_field_value(creator.address), "签约OOIN?": format_field_value(creator.has_ooin), "渠道来源": format_field_value(creator.source), "建联进度": format_field_value(creator.contact_status), "合作品牌": creator.cooperation_brands, "系统展示的带货品类": format_field_value(creator.system_categories, 'multi_select'), "实际高播放量带货品类": format_field_value(creator.actual_categories, 'multi_select'), "达人标想要货品类": format_field_value(creator.human_categories) } # 过滤掉None值 fields = {k: v for k, v in fields.items() if v is not None} if creator.record_id: # 更新现有记录 request = UpdateAppTableRecordRequest.builder() \ .app_token(APP_TOKEN) \ .table_id(TABLE_ID) \ .record_id(creator.record_id) \ .request_body(UpdateAppTableRecordRequestBody.builder().fields(fields).build()) \ .build() option = lark.RequestOption.builder().user_access_token(USER_ACCESS_TOKEN).build() response = client.bitable.v1.app_table_record.update(request, option) if response.success(): updated_count += 1 print(f"已更新记录: {creator.handle}") else: error_count += 1 print(f"更新失败: {creator.handle}, 错误: {response.msg}") else: # 创建新记录 request = CreateAppTableRecordRequest.builder() \ .app_token(APP_TOKEN) \ .table_id(TABLE_ID) \ .request_body(CreateAppTableRecordRequestBody.builder().fields(fields).build()) \ .build() option = lark.RequestOption.builder().user_access_token(USER_ACCESS_TOKEN).build() response = client.bitable.v1.app_table_record.create(request, option) if response.success(): # 获取新记录的ID并保存 record_id = response.data.record_id creator.record_id = record_id creator.save() created_count += 1 print(f"已创建记录: {creator.handle}, ID: {record_id}") else: error_count += 1 print(f"创建失败: {creator.handle}, 错误: {response.msg}") except Exception as e: error_count += 1 print(f"处理记录 {creator.handle} 时出错: {str(e)}") import traceback print(traceback.format_exc()) print("\n飞书导出完成!统计信息:") print(f"新建记录:{created_count}") print(f"更新记录:{updated_count}") print(f"错误记录:{error_count}") print(f"总处理记录数:{created_count + updated_count + error_count}") def import_from_database_to_feishu(model_name, filters=None, limit=None): """从指定数据库模型导入数据到飞书多维表格 参数: model_name (str): 模型名称,如 'Data', 'KnowledgeBase' 等 filters (dict): 过滤条件,如 {'department': '技术部'} limit (int): 限制导入的记录数量 """ # 创建client client = lark.Client.builder() \ .enable_set_token(True) \ .log_level(lark.LogLevel.DEBUG) \ .build() # 配置参数 APP_TOKEN = "XYE6bMQUOaZ5y5svj4vcWohGnmg" TABLE_ID = "tbl3oikG3F8YYtVA" USER_ACCESS_TOKEN = "u-ecM5BmzKx4uHz3sG0FouQSk1l9kxgl_3Xa00l5Ma24Jy" print(f"开始从{model_name}模型导入数据到飞书...") # 选择模型 model_mapping = { 'Data': Data, 'KnowledgeBase': KnowledgeBase, 'FeishuCreator': FeishuCreator, } model_class = model_mapping.get(model_name) if not model_class: print(f"错误: 不支持的模型名称 {model_name}") return # 构建查询 query = model_class.objects.all() # 应用过滤条件 if filters: q_objects = Q() for field, value in filters.items(): if isinstance(value, list): # 对列表值使用OR查询 q_filter = Q() for v in value: q_filter |= Q(**{field: v}) q_objects &= q_filter else: q_objects &= Q(**{field: value}) query = query.filter(q_objects) # 限制记录数量 if limit and isinstance(limit, int) and limit > 0: query = query[:limit] # 检查查询结果 total_count = query.count() if total_count == 0: print("没有符合条件的记录") return print(f"找到 {total_count} 条记录,准备导入到飞书...") # 记录统计 created_count = 0 error_count = 0 # 处理不同模型的字段映射 for index, record in enumerate(query): try: if model_name == 'Data': # Data模型到飞书字段的映射 fields = { "名称": format_field_value(record.name), "描述": format_field_value(record.desc), "类型": format_field_value(record.get_type_display()), "部门": format_field_value(record.department), "字符长度": format_field_value(record.char_length, 'number'), "文档数量": format_field_value(record.document_count, 'number'), "创建时间": format_field_value(record.create_time.strftime('%Y-%m-%d %H:%M:%S')), "更新时间": format_field_value(record.update_time.strftime('%Y-%m-%d %H:%M:%S')), "ID": format_field_value(str(record.id)), } elif model_name == 'KnowledgeBase': # KnowledgeBase模型到飞书字段的映射 fields = { "名称": format_field_value(record.name), "描述": format_field_value(record.desc), "类型": format_field_value(record.type), "部门": format_field_value(record.department), "小组": format_field_value(record.group), "文档数量": format_field_value(record.document_count, 'number'), "字符长度": format_field_value(record.char_length, 'number'), "创建时间": format_field_value(record.create_time.strftime('%Y-%m-%d %H:%M:%S')), "ID": format_field_value(str(record.id)), } else: # 对于FeishuCreator模型,使用export_to_feishu函数的逻辑 continue # 过滤掉None值 fields = {k: v for k, v in fields.items() if v is not None} # 创建新记录 request = CreateAppTableRecordRequest.builder() \ .app_token(APP_TOKEN) \ .table_id(TABLE_ID) \ .request_body(CreateAppTableRecordRequestBody.builder().fields(fields).build()) \ .build() option = lark.RequestOption.builder().user_access_token(USER_ACCESS_TOKEN).build() response = client.bitable.v1.app_table_record.create(request, option) if response.success(): created_count += 1 if created_count % 10 == 0 or created_count == total_count: print(f"已导入 {created_count}/{total_count} 条记录...") else: error_count += 1 print(f"导入第 {index+1} 条记录失败,错误: {response.msg}") except Exception as e: error_count += 1 print(f"处理第 {index+1} 条记录时出错: {str(e)}") print("\n飞书导入完成!统计信息:") print(f"成功导入记录数:{created_count}") print(f"失败记录数:{error_count}") print(f"总记录数:{total_count}") def customize_import_to_feishu(sql_query, field_mapping=None): """从自定义SQL查询结果导入数据到飞书多维表格 参数: sql_query (str): SQL查询语句 field_mapping (dict): 字段映射,如 {'db_field': 'feishu_field'} """ # 创建client client = lark.Client.builder() \ .enable_set_token(True) \ .log_level(lark.LogLevel.DEBUG) \ .build() # 配置参数 APP_TOKEN = "XYE6bMQUOaZ5y5svj4vcWohGnmg" TABLE_ID = "tbl3oikG3F8YYtVA" USER_ACCESS_TOKEN = "u-ecM5BmzKx4uHz3sG0FouQSk1l9kxgl_3Xa00l5Ma24Jy" print("开始从自定义SQL查询导入数据到飞书...") try: # 执行SQL查询 with connection.cursor() as cursor: cursor.execute(sql_query) columns = [col[0] for col in cursor.description] result_set = cursor.fetchall() # 检查查询结果 if not result_set: print("SQL查询没有返回任何结果") return print(f"查询返回 {len(result_set)} 条记录,准备导入到飞书...") # 默认字段映射:数据库字段名 -> 飞书字段名 if not field_mapping: field_mapping = {col: col for col in columns} # 记录统计 created_count = 0 error_count = 0 # 处理每条记录 for index, row in enumerate(result_set): try: # 创建记录字典 record_dict = dict(zip(columns, row)) # 将数据映射到飞书字段 fields = {} for db_field, feishu_field in field_mapping.items(): if db_field in record_dict: value = record_dict[db_field] # 处理不同类型的值 if isinstance(value, (int, float)): fields[feishu_field] = format_field_value(value, 'number') else: fields[feishu_field] = format_field_value(value) # 过滤掉None值 fields = {k: v for k, v in fields.items() if v is not None} # 创建新记录 request = CreateAppTableRecordRequest.builder() \ .app_token(APP_TOKEN) \ .table_id(TABLE_ID) \ .request_body(CreateAppTableRecordRequestBody.builder().fields(fields).build()) \ .build() option = lark.RequestOption.builder().user_access_token(USER_ACCESS_TOKEN).build() response = client.bitable.v1.app_table_record.create(request, option) if response.success(): created_count += 1 if created_count % 10 == 0 or created_count == len(result_set): print(f"已导入 {created_count}/{len(result_set)} 条记录...") else: error_count += 1 print(f"导入第 {index+1} 条记录失败,错误: {response.msg}") except Exception as e: error_count += 1 print(f"处理第 {index+1} 条记录时出错: {str(e)}") import traceback print(traceback.format_exc()) print("\n自定义SQL导入完成!统计信息:") print(f"成功导入记录数:{created_count}") print(f"失败记录数:{error_count}") print(f"总记录数:{len(result_set)}") except Exception as e: print(f"执行SQL查询时出错: {str(e)}") import traceback print(traceback.format_exc()) def print_help(): """打印帮助信息""" print("飞书数据同步工具") print("================") print("\n可用命令:") print(" 1. 从飞书导入数据到数据库:") print(" python feishu.py sync_from_feishu") print() print(" 2. 从Excel文件更新数据库记录:") print(" python feishu.py update_from_excel ") print() print(" 3. 从数据库导出数据到飞书多维表格:") print(" python feishu.py export_to_feishu [--handle 关键字] [--limit 数量]") print() print(" 4. 从特定数据库模型导入数据到飞书:") print(" python feishu.py import_from_db <模型名称> [--filters 字段1=值1,字段2=值2,...] [--limit 数量]") print(" 支持的模型: Data, KnowledgeBase, FeishuCreator") print(" 示例: python feishu.py import_from_db Data --filters type=admin,department=技术部 --limit 50") print() print(" 5. 从自定义SQL查询导入数据到飞书:") print(" python feishu.py custom_import ") print(" SQL文件中应包含SQL查询语句,及可选的字段映射JSON") print() print(" 6. 将达人信息同步到知识库:") print(" python feishu.py sync_to_kb --id | --handle | --email ") print(" 示例: python feishu.py sync_to_kb --handle tiktok_user123") print() print(" 7. 显示帮助信息:") print(" python feishu.py help") print() def sync_to_knowledge_base(creator_id=None, handle=None, email=None): """将达人信息同步到外部知识库 参数: creator_id: 达人ID handle: 达人Handle email: 达人Email 返回: tuple: (KnowledgeBase对象, 是否新创建) """ try: import json import requests import tempfile import os from datetime import datetime from django.conf import settings from user_management.models import FeishuCreator, KnowledgeBase, User, KnowledgeBaseDocument # 查找达人 query = FeishuCreator.objects.all() if creator_id: query = query.filter(id=creator_id) elif handle: query = query.filter(handle=handle) elif email: query = query.filter(email=email) creator = query.first() if not creator: logger.error(f"未找到达人: creator_id={creator_id}, handle={handle}, email={email}") return None, False # 检查达人是否有足够信息 if not creator.handle: logger.error(f"达人没有handle,无法创建知识库: {creator.id}") return None, False # 检查是否已有通过Gmail映射的知识库 gmail_kb = None if creator.email: from user_management.models import GmailTalentMapping gmail_mapping = GmailTalentMapping.objects.filter( talent_email=creator.email, is_active=True ).first() if gmail_mapping and gmail_mapping.knowledge_base: gmail_kb = gmail_mapping.knowledge_base logger.info(f"达人 {creator.handle} 已有Gmail知识库: {gmail_kb.id}") return gmail_kb, False # 获取当前用户(组长) User = get_user_model() admin_user = User.objects.filter(role='leader').first() if not admin_user: logger.error("未找到组长用户,无法创建知识库") return None, False # 生成知识库名称 kb_name = f"达人-{creator.handle}" # 检查是否已存在同名知识库 existed_kb = KnowledgeBase.objects.filter(name=kb_name).first() if existed_kb: logger.info(f"知识库 {kb_name} 已存在,ID: {existed_kb.id}") return existed_kb, False # 创建新知识库 kb = KnowledgeBase.objects.create( name=kb_name, desc=f"达人 {creator.handle} 的信息知识库", type='private', # 设置为私有知识库 user_id=admin_user.id, department=admin_user.department, document_count=0 ) logger.info(f"已创建知识库: {kb.id}") # 调用内部方法创建外部知识库 from user_management.views import KnowledgeBaseViewSet kb_viewset = KnowledgeBaseViewSet() try: external_id = kb_viewset._create_external_dataset(kb) # 保存external_id kb.external_id = external_id kb.save() logger.info(f"已创建外部知识库,ID: {external_id}") except Exception as e: logger.error(f"创建外部知识库失败: {str(e)}") # 如果外部知识库创建失败,删除本地知识库 kb.delete() return None, False # 准备达人信息文档 content = [] # 基本信息 content.append("# 达人基本信息") content.append(f"Handle: {creator.handle}") content.append(f"邮箱: {creator.email}") content.append(f"对接人: {creator.contact_person}") content.append(f"链接: {creator.tiktok_url}") content.append(f"粉丝数: {creator.fans_count}") content.append(f"GMV: {creator.gmv}") content.append(f"电话: {creator.phone}") # 账号属性 content.append("\n# 账号属性") content.append(f"账号属性: {creator.account_type}") content.append(f"报价: {creator.price_quote}") content.append(f"回复速度: {creator.response_speed}") # 合作信息 content.append("\n# 合作信息") content.append(f"合作意向: {creator.cooperation_intention}") content.append(f"支付方式: {creator.payment_method}") content.append(f"收款账号: {creator.payment_account}") content.append(f"收件地址: {creator.address}") content.append(f"签约OOIN?: {creator.has_ooin}") # 渠道和品类 content.append("\n# 渠道和品类") content.append(f"渠道来源: {creator.source}") content.append(f"建联进度: {creator.contact_status}") content.append(f"合作品牌: {creator.cooperation_brands}") content.append(f"系统展示品类: {creator.system_categories}") content.append(f"实际品类: {creator.actual_categories}") content.append(f"达人想要品类: {creator.human_categories}") # 其他信息 content.append("\n# 其他信息") content.append(f"达人base: {creator.creator_base}") content.append(f"备注: {creator.notes}") content.append(f"更新时间: {creator.updated_at.strftime('%Y-%m-%d %H:%M:%S')}") # 合并内容 full_content = "\n".join(content) # 创建临时文件 with tempfile.NamedTemporaryFile(delete=False, suffix='.txt', mode='w', encoding='utf-8') as temp: temp.write(full_content) temp_file_path = temp.name try: # 准备文档上传需要的段落数据 doc_data = { "name": f"达人信息-{creator.handle}.txt", "paragraphs": [ { "title": "达人基本资料", "content": full_content, "is_active": True, "problem_list": [] } ] } # 调用上传API upload_response = kb_viewset._call_upload_api(kb.external_id, doc_data) if upload_response and upload_response.get('code') == 200 and upload_response.get('data'): # 上传成功,保存记录到数据库 document_id = upload_response['data']['id'] doc_record = KnowledgeBaseDocument.objects.create( knowledge_base=kb, document_id=document_id, document_name=f"达人信息-{creator.handle}.txt", external_id=document_id, status='active' ) logger.info(f"文档上传成功,ID: {document_id}") # 更新知识库文档计数 kb.document_count = 1 kb.save() # 如果有邮箱,创建Gmail映射 if creator.email: from user_management.models import GmailTalentMapping GmailTalentMapping.objects.create( user=admin_user, talent_email=creator.email, knowledge_base=kb, conversation_id=f"feishu_{creator.id}", is_active=True ) logger.info(f"已创建Gmail映射: {creator.email} -> {kb.id}") logger.info(f"成功为达人 {creator.handle} 创建知识库并上传文档: {kb.id}") return kb, True else: # 上传失败,记录错误信息 error_msg = upload_response.get('message', '未知错误') if upload_response else '上传API调用失败' logger.error(f"文档上传失败: {error_msg}") # 尝试删除创建的知识库和外部知识库 try: kb_viewset._delete_external_dataset(kb.external_id) except Exception as del_err: logger.error(f"删除外部知识库失败: {str(del_err)}") kb.delete() return None, False except Exception as e: logger.error(f"文档上传过程中出错: {str(e)}") import traceback logger.error(traceback.format_exc()) # 尝试删除创建的知识库和外部知识库 try: kb_viewset._delete_external_dataset(kb.external_id) except Exception as del_err: logger.error(f"删除外部知识库失败: {str(del_err)}") kb.delete() return None, False finally: # 清理临时文件 try: os.unlink(temp_file_path) except: pass except Exception as e: logger.error(f"同步达人到知识库时出错: {str(e)}") import traceback logger.error(traceback.format_exc()) return None, False def main(): """主函数""" if len(sys.argv) < 2 or sys.argv[1] == "help": print_help() return command = sys.argv[1] if command == "sync_from_feishu": sync_from_feishu() elif command == "update_from_excel": if len(sys.argv) < 3: print("错误: 未指定Excel文件路径") print("用法: python feishu.py update_from_excel ") return excel_file_path = sys.argv[2] update_from_excel(excel_file_path) elif command == "export_to_feishu": # 可选参数处理 handle_filter = None limit = None i = 2 while i < len(sys.argv): if sys.argv[i] == "--handle" and i + 1 < len(sys.argv): handle_filter = sys.argv[i+1] i += 2 elif sys.argv[i] == "--limit" and i + 1 < len(sys.argv): try: limit = int(sys.argv[i+1]) except ValueError: print(f"警告: 无效的limit值 '{sys.argv[i+1]}',将使用默认值") i += 2 else: i += 1 # 如果提供了handle过滤器,创建自定义查询集 if handle_filter: query_set = FeishuCreator.objects.filter(handle__icontains=handle_filter) if limit: query_set = query_set[:limit] export_to_feishu(query_set) else: export_to_feishu() elif command == "import_from_db": if len(sys.argv) < 3: print("错误: 未指定数据库模型") print("用法: python feishu.py import_from_db <模型名称> [--filters 字段1=值1,字段2=值2,...] [--limit 数量]") print("支持的模型: Data, KnowledgeBase, FeishuCreator") return model_name = sys.argv[2] filters = {} limit = None # 解析额外参数 i = 3 while i < len(sys.argv): if sys.argv[i] == "--filters" and i + 1 < len(sys.argv): filter_str = sys.argv[i+1] for item in filter_str.split(','): if '=' in item: key, value = item.split('=', 1) filters[key.strip()] = value.strip() i += 2 elif sys.argv[i] == "--limit" and i + 1 < len(sys.argv): try: limit = int(sys.argv[i+1]) except ValueError: print(f"警告: 无效的limit值 '{sys.argv[i+1]}',将使用默认值") i += 2 else: i += 1 import_from_database_to_feishu(model_name, filters, limit) elif command == "custom_import": if len(sys.argv) < 3: print("错误: 未指定SQL文件路径") print("用法: python feishu.py custom_import ") return sql_file_path = sys.argv[2] # 读取SQL文件 try: with open(sql_file_path, 'r', encoding='utf-8') as f: file_content = f.read().strip() # 判断是否包含字段映射 if '-- FIELD_MAPPING:' in file_content: sql_part, mapping_part = file_content.split('-- FIELD_MAPPING:', 1) sql_query = sql_part.strip() try: field_mapping = json.loads(mapping_part.strip()) print(f"使用自定义字段映射: {field_mapping}") except json.JSONDecodeError: print(f"警告: 字段映射格式无效,将使用默认映射") field_mapping = None else: sql_query = file_content field_mapping = None # 执行导入 customize_import_to_feishu(sql_query, field_mapping) except FileNotFoundError: print(f"错误: 找不到文件 '{sql_file_path}'") except Exception as e: print(f"处理SQL文件时出错: {str(e)}") elif command == "sync_to_kb": # 同步达人信息到知识库 if len(sys.argv) < 3: print("错误: 未指定达人识别信息") print("用法: python feishu.py sync_to_kb --id | --handle | --email ") return creator_id = None handle = None email = None i = 2 while i < len(sys.argv): if sys.argv[i] == "--id" and i + 1 < len(sys.argv): creator_id = sys.argv[i+1] i += 2 elif sys.argv[i] == "--handle" and i + 1 < len(sys.argv): handle = sys.argv[i+1] i += 2 elif sys.argv[i] == "--email" and i + 1 < len(sys.argv): email = sys.argv[i+1] i += 2 else: i += 1 kb, created = sync_to_knowledge_base(creator_id, handle, email) if kb: status = "新创建" if created else "已存在" print(f"达人知识库{status}: {kb.id} ({kb.name})") print(f"文档数量: {kb.document_count}") else: print("同步失败,未能创建或找到达人知识库") else: print(f"未知命令: {command}") print_help() if __name__ == "__main__": main()