erpnew/business/tasks.py

import logging
from typing import Any, Dict, List

from celery import shared_task

from business import services as business_services

logger = logging.getLogger(__name__)


@shared_task(bind=True)
def create_purchase_order_stock_entries(
    self,
    *,
    purchase_order_id: int,
    warehouse_id: int,
    items: List[Dict[str, Any]],
    created_by_id: int | None = None,
) -> Dict[str, Any]:
    payload = business_services.create_purchase_order_stock_entries_sync(
        purchase_order_id=purchase_order_id,
        warehouse_id=warehouse_id,
        items=items,
        created_by_id=created_by_id,
    )
    payload['task_id'] = self.request.id
    return payload


@shared_task(bind=True)
def sync_external_customer_finance_scheduled(self) -> Dict[str, Any]:
    """
    定时任务：同步指定客户列表的外部财务（对账）数据。

    每天凌晨 5 点由 celery beat 触发，逐个客户调用 sync_customer_finance()。
    单个客户失败时跳过并记录错误，不影响其他客户。
    """
    from django.conf import settings
    from basic_info.models import Employee
    from business.external_finance_sync import sync_customer_finance

    customer_names = getattr(settings, 'FINANCE_SYNC_CUSTOMER_NAMES', [])
    operator_id = getattr(settings, 'HAOBUYE_FINANCE_SYNC_OPERATOR_ID', 0)

    if not customer_names:
        logger.warning('[finance_sync_scheduled] FINANCE_SYNC_CUSTOMER_NAMES 为空，跳过')
        return {'status': 'skipped', 'reason': 'empty_customer_list'}

    if not operator_id:
        logger.error('[finance_sync_scheduled] HAOBUYE_FINANCE_SYNC_OPERATOR_ID 未配置')
        return {'status': 'error', 'reason': 'missing_operator_id'}

    try:
        operator = Employee.objects.select_related('merchant').get(id=operator_id)
    except Employee.DoesNotExist:
        logger.error('[finance_sync_scheduled] 经办人不存在: id=%s', operator_id)
        return {'status': 'error', 'reason': f'operator_not_found:{operator_id}'}

    results: List[Dict[str, Any]] = []
    errors: List[Dict[str, str]] = []

    for customer_name in customer_names:
        try:
            logger.info('[finance_sync_scheduled] 开始同步客户: %s', customer_name)
            result = sync_customer_finance(
                customer_name=customer_name,
                operator=operator,
                allow_create_customer=True,
                force_update=True,
            )
            results.append({
                'customer_name': customer_name,
                'created_count': result.get('created_count', 0),
                'skipped_existing_count': result.get('skipped_existing_count', 0),
                'external_business_created_count': result.get('external_business_created_count', 0),
            })
            logger.info(
                '[finance_sync_scheduled] 客户 %s 同步完成: created=%s, skipped=%s',
                customer_name,
                result.get('created_count', 0),
                result.get('skipped_existing_count', 0),
            )
        except Exception as exc:
            logger.exception('[finance_sync_scheduled] 客户 %s 同步失败', customer_name)
            errors.append({'customer_name': customer_name, 'error': str(exc)})

    summary = {
        'status': 'completed',
        'total_customers': len(customer_names),
        'success_count': len(results),
        'error_count': len(errors),
        'results': results,
        'errors': errors,
    }
    logger.info(
        '[finance_sync_scheduled] 全部完成: success=%s, errors=%s',
        len(results),
        len(errors),
    )
    return summary



@shared_task(bind=True)
def create_sales_order_stock_entries(
    self,
    *,
    sales_order_id: int,
    warehouse_id: int,
    items: List[Dict[str, Any]],
    created_by_id: int | None = None,
) -> Dict[str, Any]:
    payload = business_services.create_sales_order_stock_entries_sync(
        sales_order_id=sales_order_id,
        warehouse_id=warehouse_id,
        items=items,
        created_by_id=created_by_id,
    )
    payload['task_id'] = self.request.id
    return payload


@shared_task(bind=True)
def create_purchase_return_order_stock_entries(
    self,
    *,
    purchase_return_order_id: int,
    warehouse_id: int,
    items: List[Dict[str, Any]],
    created_by_id: int | None = None,
) -> Dict[str, Any]:
    payload = business_services.create_purchase_return_order_stock_entries_sync(
        purchase_return_order_id=purchase_return_order_id,
        warehouse_id=warehouse_id,
        items=items,
        created_by_id=created_by_id,
    )
    payload['task_id'] = self.request.id
    return payload


@shared_task(bind=True)
def create_sales_return_order_stock_entries(
    self,
    *,
    sales_return_order_id: int,
    warehouse_id: int,
    items: List[Dict[str, Any]],
    created_by_id: int | None = None,
) -> Dict[str, Any]:
    payload = business_services.create_sales_return_order_stock_entries_sync(
        sales_return_order_id=sales_return_order_id,
        warehouse_id=warehouse_id,
        items=items,
        created_by_id=created_by_id,
    )
    payload['task_id'] = self.request.id
    return payload


@shared_task(bind=True)
def notify_business_order_speech(self, *, text: str) -> Dict[str, Any]:
    try:
        from flower.utils import play_speech

        response = play_speech(text=text, timeout_seconds=3.0)
        return {
            'status': 'sent',
            'task_id': self.request.id,
            'status_code': response.status_code,
            'raw': response.raw,
        }
    except Exception as exc:
        logger.exception('[business.tasks] 业务单据语音播报失败（已忽略）')
        return {
            'status': 'error',
            'task_id': self.request.id,
            'error': str(exc),
        }