import asyncio from typing import List, Dict, Any import concurrent # 添加这行 import concurrent.futures from app.core.crud import CRUDBase from app.schemas.crm import ( CrmCustomerCreate, CrmCustomerUpdate, CrmCustomerCreate, CrmBindInfoCreate, CrmBindInfoUpdate, ) from some_sdk.services import binder as binder_service from some_sdk.lintao_sdk.biz.by_order import get_order_relative_user from app.models.weixin import CrmCustomer, CrmBindInfo from tortoise.expressions import Subquery from tortoise.functions import Count # 导入 Count 函数 import logging logger = logging.getLogger(__name__) def split_generator(iterable, chunk_size=50): """ 将可迭代对象拆分为多个子列表 """ chunk = [] for item in iterable: chunk.append(item) if len(chunk) == chunk_size: yield chunk chunk = [] if chunk: yield chunk class CrmCustomerController(CRUDBase[CrmCustomer, CrmCustomerCreate, CrmCustomerUpdate]): def __init__(self): super().__init__(model=CrmCustomer) async def get_unbound_customers_method_four_alt(self, limit: int = 50): # 创建子查询 bound_ids_subquery = CrmBindInfo.all().values_list('customer_id', flat=True) # 使用正确的子查询语法 unbound_customers = await self.model.filter( cid__not_in=Subquery(bound_ids_subquery) ).limit(limit) return unbound_customers async def bind_crm_all_customer_by_db(self): # 定义一个函数来执行单个同步的 bind_shop_info 调用 def sync_fetch_bind_info(user_data: Dict[str, Any]): # 同步调用 binder_service all_users = [] for crm_user, user_list_from_bind in binder_service.bind_shop_info([user_data]): all_users.extend(user_list_from_bind) return all_users parsed_set = set() while True: # 获取一批未绑定的用户 ORM 对象 unbound_orm_objects = await self.get_unbound_customers_method_four_alt(limit=50) if not unbound_orm_objects: break print(f'获取到 {len(unbound_orm_objects)} 个待绑定的 ORM 对象', flush=True) expected_cids = {obj.cid for obj in unbound_orm_objects if obj.cid is not None and obj.cid not in parsed_set} if not expected_cids: break user_data_list = [await obj.to_dict() for obj in unbound_orm_objects if obj.cid not in parsed_set] [parsed_set.add(cid) for cid in expected_cids] print(f'待绑定用户数（转换后）：{len(user_data_list)}', flush=True) await self._bind_crm_all_customer_by_db(user_data_list, expected_cids, sync_fetch_bind_info) async def _bind_crm_all_customer_by_db(self, not_in_db_list: List[Dict[str, Any]], expected_cids: set, sync_fetch_bind_info): print(f'待绑定用户数（传入列表）：{len(not_in_db_list)}', flush=True) # --- 并发执行所有同步的 bind_shop_info 调用 --- # 使用 asyncio.to_thread (Python 3.9+) 或 run_in_executor 将同步函数移到线程池执行 # 限制并发线程数很重要，避免创建过多线程 max_workers = 10 # 限制线程池大小，根据系统性能调整 loop = asyncio.get_event_loop() # 创建任务列表 tasks = [] with concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) as executor: for user_data in not_in_db_list: # 将同步函数提交到线程池执行，并返回一个 Future # asyncio.run_in_executor 将 Future 包装成 awaitable 的协程 task = loop.run_in_executor(executor, sync_fetch_bind_info, user_data) tasks.append(task) # 等待所有线程池任务完成 all_bind_lists_results = await asyncio.gather(*tasks, return_exceptions=True) # 处理结果，将异常和正常结果分开 all_users_to_process = [] for result in all_bind_lists_results: if isinstance(result, Exception): logger.exception(f"Thread pool task failed: {result}") # 可以选择跳过或记录错误 else: all_users_to_process.extend(result) # --- 去重逻辑 --- unique_users_to_process = {} for user in all_users_to_process: customer_id = user.get("customer_id", 0) tenant_name = user.get("tenant_name", 0) platform = user.get("platform", 0) key = f"{customer_id}_{tenant_name}_{platform}" if key not in unique_users_to_process: unique_users_to_process[key] = user print(f"去重后待处理用户数: {len(unique_users_to_process)}", flush=True) # --- 并发处理去重后的用户 --- # 这部分可以保持原有逻辑，因为它已经是并发的了 # 但要注意，如果同时有多个任务尝试插入相同的 customer_id，可能会有并发问题 # 可以考虑使用数据库的 INSERT IGNORE 或 ON DUPLICATE KEY UPDATE 等特性 async def process_user(user_obj: Dict[str, Any]): try: user_type = user_obj.get("type", 0) tenant_name = user_obj.get("tenant_name", 0) customer_id = user_obj.get("customer_id", 0) platform = user_obj.get("platform", 0) if user_type == 'staff': print(f'绑定员工：{user_obj}', flush=True) else: print(f'绑定客户：{user_obj}, Customer ID: {customer_id}', flush=True) # --- 检查数据库中是否已存在绑定 --- existing_bind = await CrmBindInfo.filter(customer_id=customer_id, platform=platform, tenant_name=tenant_name).first() if not existing_bind: bind_instance = CrmBindInfo.create_bind(user_obj) await bind_instance.save() print(f"客户 {customer_id} 绑定成功", flush=True) else: print(f"客户 {customer_id} 已存在绑定", flush=True) logger.warning(f"Attempted to bind customer_id {customer_id} which already exists in CrmBindInfo.") except Exception as e: logger.exception(e) print(f"处理用户 {user_obj.get('customer_id', 'unknown')} 时出错: {e}", flush=True) max_concurrent_db_tasks = 10 # 可以独立控制数据库操作的并发数 semaphore = asyncio.Semaphore(max_concurrent_db_tasks) async def process_user_with_semaphore(user_obj): async with semaphore: return await process_user(user_obj) tasks = [process_user_with_semaphore(user_obj) for user_obj in unique_users_to_process.values()] await asyncio.gather(*tasks, return_exceptions=True) print("当前批次所有用户绑定任务完成", flush=True) async def bind_crm_all_customer(self): """ 绑定CRM客户和微信用户 """ await self.bind_crm_all_customer_by_db() for shop in binder_service.iter_shop(): shopId = shop.get("shopId", "") shopName = shop.get("shopName", "") print(f"开始处理店铺：{shopName}", flush=True) not_in_db_list = [] shop_info = { 'shopId': shopId, 'shopName': shopName, } is_done = False user_list_iter = binder_service.iter_list_trade_user(binder_service.xy_client, shopId=shopId) for user_list in split_generator(user_list_iter): if is_done: break # 提取所有待检查的 cid all_cids = [str(user.get("cid")) for user in user_list] # print(user_list[0]) # break # 只查询 user_list 中存在的 cid in_objs = await self.model.filter(cid__in=all_cids) # 提取已存在的 cid in_db_cid_set = {user.cid for user in in_objs} print(f'已处理用户数：{len(in_db_cid_set)}', flush=True) # 筛选待处理用户 not_in_db_list = [user for user in user_list if user.get("cid") not in in_db_cid_set] for user in not_in_db_list: id = user.pop("id", None) # print('user', user) # if not not_in_db_list: # is_done = True # break print(f'待处理用户数：{len(not_in_db_list)}', flush=True) if not_in_db_list: model_list = [self.model(**CrmCustomerCreate(**user).model_dump(exclude_unset=True)) for user in not_in_db_list] await self.model.bulk_create(model_list) if not not_in_db_list: print(f'店铺：{shopName}，无待处理用户', flush=True) continue await self.bind_crm_all_customer_by_db() async def bind_user(self, bind_info: CrmBindInfoCreate): """ 绑定用户 """ bind_info_dict = bind_info.model_dump(exclude_unset=True) platform = bind_info_dict.get("platform", None) platform_id = bind_info_dict.get("platform_id", None) bind_info_dict.pop("id", None) existing_bind = await CrmBindInfo.filter(platform=platform, platform_id=platform_id).first() if existing_bind: await CrmBindInfo.filter(id=existing_bind.id).update(**bind_info_dict) else: obj = CrmBindInfo(**bind_info_dict) await obj.save() return {"message": "用户绑定成功"} crm_customer_controller = CrmCustomerController()