forked from erp-dev/erp
feat: correct first
This commit is contained in:
841
business/external_finance_sync.py
Normal file
841
business/external_finance_sync.py
Normal file
@@ -0,0 +1,841 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import logging
|
||||
from dataclasses import dataclass
|
||||
from datetime import date, datetime
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from typing import Any, Callable, Iterable
|
||||
|
||||
import requests
|
||||
from django.conf import settings
|
||||
from django.db import transaction
|
||||
|
||||
from basic_info import models as basic_models
|
||||
from . import models as business_models
|
||||
from . import services as business_services
|
||||
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
class ExternalFinanceSyncError(RuntimeError):
|
||||
pass
|
||||
|
||||
|
||||
class ExternalFinanceSyncConflictError(ExternalFinanceSyncError):
|
||||
pass
|
||||
|
||||
|
||||
@dataclass(slots=True)
|
||||
class ExternalCustomerRef:
|
||||
customer_id: str
|
||||
customer_name: str
|
||||
|
||||
|
||||
class HaoBuYeFinanceClient:
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
base_url: str | None = None,
|
||||
authorization: str | None = None,
|
||||
timeout_seconds: float | None = None,
|
||||
session: requests.Session | None = None,
|
||||
):
|
||||
self.base_url = str(base_url or getattr(settings, 'HAOBUYE_API_BASE_URL', '') or '').rstrip('/')
|
||||
self.authorization = str(
|
||||
authorization or getattr(settings, 'HAOBUYE_API_AUTHORIZATION', '') or ''
|
||||
).strip()
|
||||
self.timeout_seconds = float(
|
||||
timeout_seconds
|
||||
if timeout_seconds is not None
|
||||
else getattr(settings, 'HAOBUYE_API_TIMEOUT_SECONDS', 30.0)
|
||||
)
|
||||
self.session = session or requests.Session()
|
||||
if not self.base_url:
|
||||
raise ExternalFinanceSyncError('未配置 HAOBUYE_API_BASE_URL')
|
||||
if not self.authorization:
|
||||
raise ExternalFinanceSyncError('未配置 HAOBUYE_API_AUTHORIZATION')
|
||||
|
||||
def list_customers(self, *, cursor_id: str | None = None) -> dict[str, Any]:
|
||||
params: dict[str, str] = {}
|
||||
if cursor_id:
|
||||
params['cursor_id'] = str(cursor_id).strip()
|
||||
payload = self._request_json('/api/v1/customers', params=params)
|
||||
if not isinstance(payload.get('customers', []), list):
|
||||
raise ExternalFinanceSyncError('外部 customers 接口返回格式非法：customers 不是数组')
|
||||
return payload
|
||||
|
||||
def fetch_customer_finance(
|
||||
self,
|
||||
*,
|
||||
customer_name: str,
|
||||
include_adjustments: bool = True,
|
||||
include_cash_movement: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
normalized_name = str(customer_name or '').strip()
|
||||
if not normalized_name:
|
||||
raise ExternalFinanceSyncError('customer_name 不能为空')
|
||||
customer_name_b64 = base64.urlsafe_b64encode(normalized_name.encode('utf-8')).decode('ascii').rstrip('=')
|
||||
payload = self._request_json(
|
||||
'/api/v1/finance/by-customer',
|
||||
params={
|
||||
'customer_name_b64': customer_name_b64,
|
||||
'record_types': 'receipt,refund,sale_discount',
|
||||
'include_cash_movement': str(include_cash_movement).lower(),
|
||||
'include_adjustments': str(include_adjustments).lower(),
|
||||
},
|
||||
)
|
||||
return payload
|
||||
|
||||
def fetch_i_sale_by_customer(
|
||||
self,
|
||||
*,
|
||||
customer_id: str,
|
||||
category: str,
|
||||
) -> dict[str, Any]:
|
||||
normalized_customer_id = str(customer_id or '').strip()
|
||||
normalized_category = str(category or '').strip()
|
||||
if not normalized_customer_id:
|
||||
raise ExternalFinanceSyncError('customer_id 不能为空')
|
||||
if normalized_category not in {
|
||||
business_models.ExternalCustomerStatementCategoryEnum.SALE,
|
||||
business_models.ExternalCustomerStatementCategoryEnum.SALE_RETURN,
|
||||
}:
|
||||
raise ExternalFinanceSyncError(f'不支持的 i_sale category: {normalized_category}')
|
||||
return self._request_json(
|
||||
'/api/v1/i-sale/by-customer',
|
||||
params={
|
||||
'customer_id': normalized_customer_id,
|
||||
'category': normalized_category,
|
||||
},
|
||||
)
|
||||
|
||||
def _request_json(self, path: str, *, params: dict[str, str] | None = None) -> dict[str, Any]:
|
||||
url = f'{self.base_url}{path}'
|
||||
try:
|
||||
response = self.session.get(
|
||||
url,
|
||||
headers={'Authorization': self.authorization},
|
||||
params=params,
|
||||
timeout=self.timeout_seconds,
|
||||
)
|
||||
except requests.RequestException as exc:
|
||||
raise ExternalFinanceSyncError(f'请求外部接口失败: {exc}') from exc
|
||||
if response.status_code >= 400:
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError:
|
||||
payload = {}
|
||||
message = payload.get('message') or payload.get('error') or f'外部接口异常状态码: {response.status_code}'
|
||||
raise ExternalFinanceSyncError(str(message))
|
||||
try:
|
||||
payload = response.json()
|
||||
except ValueError as exc:
|
||||
raise ExternalFinanceSyncError('外部接口返回格式非法：非 JSON object') from exc
|
||||
if not isinstance(payload, dict):
|
||||
raise ExternalFinanceSyncError('外部接口返回格式非法：非 JSON object')
|
||||
return payload
|
||||
|
||||
|
||||
def sync_customer_finance(
|
||||
*,
|
||||
customer_name: str,
|
||||
operator: basic_models.Employee,
|
||||
client: HaoBuYeFinanceClient | None = None,
|
||||
allow_create_customer: bool = True,
|
||||
dry_run: bool = False,
|
||||
force_update: bool = False,
|
||||
include_adjustments: bool = True,
|
||||
include_cash_movement: bool = True,
|
||||
progress_callback: Callable[[str], None] | None = None,
|
||||
progress_every: int = 200,
|
||||
) -> dict[str, Any]:
|
||||
merchant = _ensure_active_operator(operator)
|
||||
_emit_progress(progress_callback, f'开始同步客户: {customer_name}')
|
||||
finance_client = client or HaoBuYeFinanceClient()
|
||||
payload = finance_client.fetch_customer_finance(
|
||||
customer_name=customer_name,
|
||||
include_adjustments=include_adjustments,
|
||||
include_cash_movement=include_cash_movement,
|
||||
)
|
||||
|
||||
status_value = str(payload.get('status') or '').strip()
|
||||
normalized_name = str(payload.get('customer_name') or customer_name or '').strip()
|
||||
external_customer_id = _extract_external_customer_id(payload)
|
||||
|
||||
summary: dict[str, Any] = {
|
||||
'customer_name': normalized_name,
|
||||
'external_customer_id': external_customer_id,
|
||||
'status': status_value or 'active',
|
||||
'receipts_seen': 0,
|
||||
'refunds_seen': 0,
|
||||
'sales_seen': 0,
|
||||
'sale_returns_seen': 0,
|
||||
'created_count': 0,
|
||||
'skipped_existing_count': 0,
|
||||
'skipped_zero_settlement_count': 0,
|
||||
'external_business_created_count': 0,
|
||||
'external_business_skipped_existing_count': 0,
|
||||
'dry_run_count': 0,
|
||||
'created_receipt_ids': [],
|
||||
'created_external_business_ids': [],
|
||||
'skipped_external_ids': [],
|
||||
}
|
||||
|
||||
if status_value == 'not_found':
|
||||
return summary
|
||||
|
||||
customer = _find_or_create_customer(
|
||||
merchant=merchant,
|
||||
customer_name=normalized_name,
|
||||
external_customer_id=external_customer_id,
|
||||
operator=operator,
|
||||
allow_create_customer=allow_create_customer,
|
||||
dry_run=dry_run,
|
||||
)
|
||||
|
||||
receipts = payload.get('receipts') or []
|
||||
refunds = payload.get('refunds') or []
|
||||
if not isinstance(receipts, list) or not isinstance(refunds, list):
|
||||
raise ExternalFinanceSyncError('外部 finance 接口返回格式非法：receipts/refunds 不是数组')
|
||||
|
||||
summary['receipts_seen'] = len(receipts)
|
||||
summary['refunds_seen'] = len(refunds)
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
f'财务记录已获取: receipts={summary["receipts_seen"]}, refunds={summary["refunds_seen"]}',
|
||||
)
|
||||
|
||||
for index, record in enumerate(receipts, start=1):
|
||||
result = _sync_external_receipt_record(
|
||||
merchant=merchant,
|
||||
customer=customer,
|
||||
operator=operator,
|
||||
record=record,
|
||||
record_kind='receipt',
|
||||
dry_run=dry_run,
|
||||
)
|
||||
_merge_sync_result(summary, result)
|
||||
if index == 1 or index % progress_every == 0 or index == len(receipts):
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
f'收款同步进度: {index}/{len(receipts)}',
|
||||
)
|
||||
|
||||
# NOTE: refunds (XT% 退款记录) 不再创建 ReceiptOrder。
|
||||
# 退货的减欠效果完全由 ExternalCustomerStatementOrder(category=sale_return) 承担。
|
||||
# 之前的实现会把 refund 同时写成 ReceiptOrder，导致退货金额被双重计入，
|
||||
# 与 ERP 对账结果不一致（详见 docs/DEBT_ANALYSIS_SUMMARY_2026-05-18.md 第四节）。
|
||||
if refunds:
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
f'跳过 refunds 桶（{len(refunds)} 条），退货效果由 sale_return 业务依据承担',
|
||||
)
|
||||
|
||||
if external_customer_id:
|
||||
_emit_progress(progress_callback, f'开始同步外部业务依据: customer_id={external_customer_id}, category=sale')
|
||||
sales_result = sync_customer_external_statement_orders(
|
||||
customer=customer,
|
||||
external_customer_id=external_customer_id,
|
||||
operator=operator,
|
||||
client=finance_client,
|
||||
category=business_models.ExternalCustomerStatementCategoryEnum.SALE,
|
||||
dry_run=dry_run,
|
||||
force_update=force_update,
|
||||
progress_callback=progress_callback,
|
||||
progress_every=progress_every,
|
||||
)
|
||||
_merge_external_business_sync_result(summary, sales_result)
|
||||
|
||||
_emit_progress(progress_callback, f'开始同步外部业务依据: customer_id={external_customer_id}, category=sale_return')
|
||||
sale_return_result = sync_customer_external_statement_orders(
|
||||
customer=customer,
|
||||
external_customer_id=external_customer_id,
|
||||
operator=operator,
|
||||
client=finance_client,
|
||||
category=business_models.ExternalCustomerStatementCategoryEnum.SALE_RETURN,
|
||||
dry_run=dry_run,
|
||||
force_update=force_update,
|
||||
progress_callback=progress_callback,
|
||||
progress_every=progress_every,
|
||||
)
|
||||
_merge_external_business_sync_result(summary, sale_return_result)
|
||||
|
||||
# 处理 sale_discount 数据：从 F_Skd 的 XS% 行获取真实的销售折扣 ZkJinE，
|
||||
# 更新到对应的 ExternalCustomerStatementOrder.zk_amount 中。
|
||||
# I_Sale 表的 ZkJinE 始终为 0，真实折扣只存在于 F_Skd 表。
|
||||
sale_discounts = payload.get('sale_discounts') or []
|
||||
if sale_discounts and isinstance(sale_discounts, list) and customer:
|
||||
_apply_sale_discount_zk(
|
||||
merchant=merchant,
|
||||
customer=customer,
|
||||
sale_discounts=sale_discounts,
|
||||
progress_callback=progress_callback,
|
||||
)
|
||||
|
||||
_emit_progress(progress_callback, f'客户同步完成: {customer_name}')
|
||||
return summary
|
||||
|
||||
|
||||
def sync_customer_external_statement_orders(
|
||||
*,
|
||||
customer: basic_models.Customer,
|
||||
external_customer_id: str,
|
||||
operator: basic_models.Employee,
|
||||
client: HaoBuYeFinanceClient | None = None,
|
||||
category: str,
|
||||
dry_run: bool = False,
|
||||
force_update: bool = False,
|
||||
progress_callback: Callable[[str], None] | None = None,
|
||||
progress_every: int = 200,
|
||||
) -> dict[str, Any]:
|
||||
merchant = _ensure_active_operator(operator)
|
||||
if customer.merchant_id != merchant.id:
|
||||
raise ExternalFinanceSyncError('customer 与 operator 不属于同一 merchant')
|
||||
finance_client = client or HaoBuYeFinanceClient()
|
||||
payload = finance_client.fetch_i_sale_by_customer(customer_id=external_customer_id, category=category)
|
||||
status_value = str(payload.get('status') or '').strip()
|
||||
records = payload.get('records') or []
|
||||
if not isinstance(records, list):
|
||||
raise ExternalFinanceSyncError('外部 i_sale by customer 接口返回格式非法：records 不是数组')
|
||||
|
||||
summary = {
|
||||
'category': category,
|
||||
'seen_count': len(records),
|
||||
'created_count': 0,
|
||||
'skipped_existing_count': 0,
|
||||
'created_ids': [],
|
||||
}
|
||||
if status_value == 'not_found':
|
||||
_emit_progress(progress_callback, f'外部业务依据未命中: category={category}, customer_id={external_customer_id}')
|
||||
return summary
|
||||
|
||||
grouped_records = _group_external_i_sale_records(records)
|
||||
summary['seen_count'] = len(grouped_records)
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
f'外部业务依据已获取: category={category}, raw_records={len(records)}, grouped_orders={summary["seen_count"]}',
|
||||
)
|
||||
for index, (external_source_id, grouped) in enumerate(grouped_records.items(), start=1):
|
||||
result = _sync_external_statement_order_group(
|
||||
merchant=merchant,
|
||||
customer=customer,
|
||||
external_customer_id=external_customer_id,
|
||||
external_source_id=external_source_id,
|
||||
category=category,
|
||||
grouped_records=grouped,
|
||||
dry_run=dry_run,
|
||||
force_update=force_update,
|
||||
)
|
||||
summary['created_count'] += result['created_count']
|
||||
summary['skipped_existing_count'] += result['skipped_existing_count']
|
||||
summary['created_ids'].extend(result['created_ids'])
|
||||
if index == 1 or index % progress_every == 0 or index == len(grouped_records):
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
(
|
||||
f'外部业务依据同步进度: category={category}, {index}/{len(grouped_records)}, '
|
||||
f'created={summary["created_count"]}, skipped={summary["skipped_existing_count"]}'
|
||||
),
|
||||
)
|
||||
return summary
|
||||
|
||||
|
||||
def sync_customer_finance_batch(
|
||||
*,
|
||||
operator: basic_models.Employee,
|
||||
cursor_id: str | None = None,
|
||||
client: HaoBuYeFinanceClient | None = None,
|
||||
allow_create_customer: bool = True,
|
||||
dry_run: bool = False,
|
||||
include_adjustments: bool = True,
|
||||
include_cash_movement: bool = True,
|
||||
) -> dict[str, Any]:
|
||||
_ensure_active_operator(operator)
|
||||
finance_client = client or HaoBuYeFinanceClient()
|
||||
customers_payload = finance_client.list_customers(cursor_id=cursor_id)
|
||||
raw_customers = customers_payload.get('customers') or []
|
||||
customer_refs = [
|
||||
ExternalCustomerRef(
|
||||
customer_id=str(item.get('customer_id') or '').strip(),
|
||||
customer_name=str(item.get('customer_name') or '').strip(),
|
||||
)
|
||||
for item in raw_customers
|
||||
if str(item.get('customer_name') or '').strip()
|
||||
]
|
||||
|
||||
results = []
|
||||
for customer_ref in customer_refs:
|
||||
results.append(
|
||||
sync_customer_finance(
|
||||
customer_name=customer_ref.customer_name,
|
||||
operator=operator,
|
||||
client=finance_client,
|
||||
allow_create_customer=allow_create_customer,
|
||||
dry_run=dry_run,
|
||||
include_adjustments=include_adjustments,
|
||||
include_cash_movement=include_cash_movement,
|
||||
)
|
||||
)
|
||||
|
||||
return {
|
||||
'mode': 'batch_customer_finance',
|
||||
'cursor_id': customers_payload.get('cursor_id'),
|
||||
'next_cursor_id': customers_payload.get('next_cursor_id'),
|
||||
'count': len(customer_refs),
|
||||
'results': results,
|
||||
}
|
||||
|
||||
|
||||
def _ensure_active_operator(operator: basic_models.Employee) -> basic_models.Merchant:
|
||||
if not operator:
|
||||
raise ExternalFinanceSyncError('operator 不能为空')
|
||||
if operator.status != basic_models.EmployeeStatusEnum.ACTIVE:
|
||||
raise ExternalFinanceSyncError('operator 必须为在职员工')
|
||||
if not operator.merchant_id:
|
||||
raise ExternalFinanceSyncError('operator 未绑定 merchant')
|
||||
return operator.merchant
|
||||
|
||||
|
||||
def _extract_external_customer_id(payload: dict[str, Any]) -> str:
|
||||
customer_ids = payload.get('customer_ids') or []
|
||||
if isinstance(customer_ids, list) and customer_ids:
|
||||
return str(customer_ids[0] or '').strip()
|
||||
return ''
|
||||
|
||||
|
||||
def _find_or_create_customer(
|
||||
*,
|
||||
merchant: basic_models.Merchant,
|
||||
customer_name: str,
|
||||
external_customer_id: str,
|
||||
operator: basic_models.Employee,
|
||||
allow_create_customer: bool,
|
||||
dry_run: bool,
|
||||
):
|
||||
customer = basic_models.Customer.objects.filter(
|
||||
merchant=merchant,
|
||||
name=customer_name,
|
||||
).order_by('id').first()
|
||||
if customer:
|
||||
return customer
|
||||
if not allow_create_customer:
|
||||
raise ExternalFinanceSyncError(f'客户不存在且未允许自动创建: {customer_name}')
|
||||
if dry_run:
|
||||
return basic_models.Customer(
|
||||
merchant=merchant,
|
||||
name=customer_name,
|
||||
created_by=operator,
|
||||
description=f'外部客户ID: {external_customer_id}' if external_customer_id else '',
|
||||
)
|
||||
return basic_models.Customer.objects.create(
|
||||
merchant=merchant,
|
||||
name=customer_name,
|
||||
created_by=operator,
|
||||
description=f'外部客户ID: {external_customer_id}' if external_customer_id else '',
|
||||
)
|
||||
|
||||
|
||||
def _sync_external_receipt_record(
|
||||
*,
|
||||
merchant: basic_models.Merchant,
|
||||
customer,
|
||||
operator: basic_models.Employee,
|
||||
record: dict[str, Any],
|
||||
record_kind: str,
|
||||
dry_run: bool,
|
||||
) -> dict[str, Any]:
|
||||
if not isinstance(record, dict):
|
||||
raise ExternalFinanceSyncError('外部 finance record 不是对象')
|
||||
|
||||
normalized = _normalize_external_receipt_record(record=record, record_kind=record_kind)
|
||||
existing = business_models.ReceiptOrder.objects.filter(
|
||||
merchant=merchant,
|
||||
external_source_id=normalized['external_source_id'],
|
||||
is_external_source=True,
|
||||
).select_related('customer').first()
|
||||
|
||||
if existing:
|
||||
if not _is_equivalent_receipt_order(existing=existing, customer=customer, normalized=normalized):
|
||||
raise ExternalFinanceSyncConflictError(
|
||||
f'外部记录 {normalized["external_source_id"]} 已存在，但本地字段与外部不一致'
|
||||
)
|
||||
return {
|
||||
'created_count': 0,
|
||||
'skipped_existing_count': 1,
|
||||
'skipped_zero_settlement_count': 0,
|
||||
'dry_run_count': 0,
|
||||
'created_receipt_ids': [],
|
||||
'skipped_external_ids': [normalized['external_source_id']],
|
||||
}
|
||||
|
||||
if dry_run:
|
||||
return {
|
||||
'created_count': 0,
|
||||
'skipped_existing_count': 0,
|
||||
'skipped_zero_settlement_count': 0,
|
||||
'dry_run_count': 1,
|
||||
'created_receipt_ids': [],
|
||||
'skipped_external_ids': [],
|
||||
}
|
||||
|
||||
with transaction.atomic():
|
||||
order = business_models.ReceiptOrder.objects.create(
|
||||
merchant=merchant,
|
||||
customer=customer,
|
||||
receipt_date=normalized['receipt_date'],
|
||||
amount=normalized['amount'],
|
||||
discount_amount=normalized['discount_amount'],
|
||||
bank_account=None,
|
||||
markup=normalized['markup'],
|
||||
operator=operator,
|
||||
status=business_models.ReceiptOrderStatusEnum.PENDING,
|
||||
is_external_source=True,
|
||||
external_source_id=normalized['external_source_id'],
|
||||
remarks=normalized['remarks'],
|
||||
)
|
||||
approved = business_services.review_receipt_order(
|
||||
receipt_order=order,
|
||||
target_status=business_models.ReceiptOrderStatusEnum.APPROVED,
|
||||
reviewed_by=operator,
|
||||
)
|
||||
|
||||
return {
|
||||
'created_count': 1,
|
||||
'skipped_existing_count': 0,
|
||||
'skipped_zero_settlement_count': 0,
|
||||
'dry_run_count': 0,
|
||||
'created_receipt_ids': [approved.id],
|
||||
'skipped_external_ids': [],
|
||||
}
|
||||
|
||||
|
||||
def _normalize_external_receipt_record(*, record: dict[str, Any], record_kind: str) -> dict[str, Any]:
|
||||
external_source_id = str(record.get('BianHaoID') or '').strip()
|
||||
if not external_source_id:
|
||||
raise ExternalFinanceSyncError('外部 finance record 缺少 BianHaoID')
|
||||
|
||||
customer_name = str(record.get('KhName') or '').strip()
|
||||
markup = str(record.get('JieSunFS') or '').strip() or None
|
||||
receipt_date = _parse_external_date(record.get('RiQi') or record.get('KdRiQi'))
|
||||
if record_kind == 'receipt':
|
||||
amount = _to_decimal(record.get('FkJinE'), field_name='FkJinE')
|
||||
discount_amount = _to_decimal(record.get('ZkJinE'), field_name='ZkJinE', default=Decimal('0'))
|
||||
elif record_kind == 'refund':
|
||||
amount = _to_decimal(record.get('YfJinE'), field_name='YfJinE')
|
||||
discount_amount = Decimal('0')
|
||||
else:
|
||||
raise ExternalFinanceSyncError(f'不支持的 record_kind: {record_kind}')
|
||||
|
||||
return {
|
||||
'external_source_id': external_source_id,
|
||||
'receipt_date': receipt_date,
|
||||
'amount': amount,
|
||||
'discount_amount': discount_amount,
|
||||
'settlement_amount': amount + discount_amount,
|
||||
'customer_name': customer_name,
|
||||
'markup': markup,
|
||||
'remarks': _build_external_remarks(record=record, record_kind=record_kind),
|
||||
}
|
||||
|
||||
|
||||
def _build_external_remarks(*, record: dict[str, Any], record_kind: str) -> str:
|
||||
lines = [f'外部财务同步: {record_kind}']
|
||||
external_customer_id = str(record.get('KhID') or '').strip()
|
||||
if external_customer_id:
|
||||
lines.append(f'外部客户ID: {external_customer_id}')
|
||||
summary = str(record.get('ZhaiYao') or '').strip()
|
||||
if summary:
|
||||
lines.append(f'摘要: {summary}')
|
||||
note = str(record.get('BeiZhu') or '').strip()
|
||||
if note:
|
||||
lines.append(f'备注: {note}')
|
||||
return '\n'.join(lines)
|
||||
|
||||
|
||||
def _parse_external_date(value: Any) -> date:
|
||||
text = str(value or '').strip()
|
||||
if not text:
|
||||
raise ExternalFinanceSyncError('外部 finance record 缺少日期字段')
|
||||
try:
|
||||
parsed = datetime.fromisoformat(text.replace('Z', '+00:00'))
|
||||
except ValueError as exc:
|
||||
raise ExternalFinanceSyncError(f'外部日期格式非法: {text}') from exc
|
||||
return parsed.date()
|
||||
|
||||
|
||||
def _to_decimal(value: Any, *, field_name: str, default: Decimal | None = None) -> Decimal:
|
||||
if value in (None, ''):
|
||||
if default is not None:
|
||||
return default
|
||||
raise ExternalFinanceSyncError(f'外部字段 {field_name} 不能为空')
|
||||
try:
|
||||
return Decimal(str(value))
|
||||
except (InvalidOperation, TypeError, ValueError) as exc:
|
||||
raise ExternalFinanceSyncError(f'外部字段 {field_name} 不是合法数字: {value!r}') from exc
|
||||
|
||||
|
||||
def _is_equivalent_receipt_order(*, existing: business_models.ReceiptOrder, customer, normalized: dict[str, Any]) -> bool:
|
||||
customer_id = getattr(customer, 'id', None)
|
||||
return all(
|
||||
[
|
||||
existing.customer_id == customer_id,
|
||||
existing.receipt_date == normalized['receipt_date'],
|
||||
existing.amount == normalized['amount'],
|
||||
existing.discount_amount == normalized['discount_amount'],
|
||||
existing.status == business_models.ReceiptOrderStatusEnum.APPROVED,
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _merge_sync_result(summary: dict[str, Any], result: dict[str, Any]) -> None:
|
||||
summary['created_count'] += int(result.get('created_count') or 0)
|
||||
summary['skipped_existing_count'] += int(result.get('skipped_existing_count') or 0)
|
||||
summary['skipped_zero_settlement_count'] += int(result.get('skipped_zero_settlement_count') or 0)
|
||||
summary['dry_run_count'] += int(result.get('dry_run_count') or 0)
|
||||
summary['created_receipt_ids'].extend(result.get('created_receipt_ids') or [])
|
||||
summary['skipped_external_ids'].extend(result.get('skipped_external_ids') or [])
|
||||
|
||||
|
||||
def _merge_external_business_sync_result(summary: dict[str, Any], result: dict[str, Any]) -> None:
|
||||
category = result.get('category')
|
||||
seen_count = int(result.get('seen_count') or 0)
|
||||
if category == business_models.ExternalCustomerStatementCategoryEnum.SALE:
|
||||
summary['sales_seen'] += seen_count
|
||||
elif category == business_models.ExternalCustomerStatementCategoryEnum.SALE_RETURN:
|
||||
summary['sale_returns_seen'] += seen_count
|
||||
summary['external_business_created_count'] += int(result.get('created_count') or 0)
|
||||
summary['external_business_skipped_existing_count'] += int(result.get('skipped_existing_count') or 0)
|
||||
summary['created_external_business_ids'].extend(result.get('created_ids') or [])
|
||||
|
||||
|
||||
def _group_external_i_sale_records(records: list[dict[str, Any]]) -> dict[str, list[dict[str, Any]]]:
|
||||
grouped: dict[str, list[dict[str, Any]]] = {}
|
||||
for record in records:
|
||||
if not isinstance(record, dict):
|
||||
raise ExternalFinanceSyncError('外部 i_sale record 不是对象')
|
||||
external_source_id = str(record.get('BianHaoID') or '').strip()
|
||||
if not external_source_id:
|
||||
raise ExternalFinanceSyncError('外部 i_sale record 缺少 BianHaoID')
|
||||
grouped.setdefault(external_source_id, []).append(record)
|
||||
return grouped
|
||||
|
||||
|
||||
def _sync_external_statement_order_group(
|
||||
*,
|
||||
merchant: basic_models.Merchant,
|
||||
customer: basic_models.Customer,
|
||||
external_customer_id: str,
|
||||
external_source_id: str,
|
||||
category: str,
|
||||
grouped_records: list[dict[str, Any]],
|
||||
dry_run: bool,
|
||||
force_update: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
normalized = _normalize_external_statement_order_group(
|
||||
customer=customer,
|
||||
external_customer_id=external_customer_id,
|
||||
category=category,
|
||||
external_source_id=external_source_id,
|
||||
grouped_records=grouped_records,
|
||||
)
|
||||
existing = business_models.ExternalCustomerStatementOrder.objects.filter(
|
||||
merchant=merchant,
|
||||
category=category,
|
||||
external_source_id=external_source_id,
|
||||
).first()
|
||||
if existing:
|
||||
if _is_equivalent_external_statement_order(existing=existing, normalized=normalized):
|
||||
return {'created_count': 0, 'skipped_existing_count': 1, 'updated_count': 0, 'created_ids': []}
|
||||
if not force_update:
|
||||
raise ExternalFinanceSyncConflictError(
|
||||
f'外部业务记录 {external_source_id} 已存在，但本地字段与外部不一致'
|
||||
)
|
||||
# force_update: 用新数据覆盖已有记录
|
||||
update_fields = [
|
||||
'total_amount', 'sf_amount', 'zk_amount', 'occurred_at', 'recorded_at',
|
||||
'settlement_method', 'remarks', 'items_payload', 'extra_payload',
|
||||
]
|
||||
for field in update_fields:
|
||||
if field in normalized:
|
||||
setattr(existing, field, normalized[field])
|
||||
existing.save(update_fields=[*update_fields, 'updated_at'])
|
||||
return {'created_count': 0, 'skipped_existing_count': 0, 'updated_count': 1, 'created_ids': []}
|
||||
|
||||
if dry_run:
|
||||
return {'created_count': 0, 'skipped_existing_count': 0, 'updated_count': 0, 'created_ids': []}
|
||||
|
||||
created = business_models.ExternalCustomerStatementOrder.objects.create(**normalized)
|
||||
return {'created_count': 1, 'skipped_existing_count': 0, 'updated_count': 0, 'created_ids': [created.id]}
|
||||
|
||||
|
||||
def _normalize_external_statement_order_group(
|
||||
*,
|
||||
customer: basic_models.Customer,
|
||||
external_customer_id: str,
|
||||
category: str,
|
||||
external_source_id: str,
|
||||
grouped_records: list[dict[str, Any]],
|
||||
) -> dict[str, Any]:
|
||||
first_record = grouped_records[0]
|
||||
occurred_at = _parse_external_date(first_record.get('RiQi') or first_record.get('KdRiQi'))
|
||||
recorded_at = _parse_external_datetime(first_record.get('KdRiQi'))
|
||||
remarks_lines = []
|
||||
items_payload = []
|
||||
total_amount = Decimal('0')
|
||||
sf_amount = Decimal('0')
|
||||
zk_amount = Decimal('0')
|
||||
|
||||
# SfJinE 是单据头级别字段（每行重复），只取首条记录的值（按 BianHaoID 去重）
|
||||
sf_amount = abs(_to_decimal(first_record.get('SfJinE'), field_name='SfJinE', default=Decimal('0')))
|
||||
# ZkJinE 不去重，每行独立求和
|
||||
|
||||
for record in grouped_records:
|
||||
# JinE 不取绝对值：同一订单内可能有负数明细行（如冲减行），应保留原始正负
|
||||
# 最终 total_amount 取绝对值（退货单整体为负，销售单整体为正）
|
||||
amount_value = _to_decimal(record.get('JinE'), field_name='JinE', default=Decimal('0'))
|
||||
price_value = abs(_to_decimal(record.get('DanJia'), field_name='DanJia', default=Decimal('0')))
|
||||
quantity_value = abs(_to_decimal(record.get('ShuLiang'), field_name='ShuLiang', default=Decimal('0')))
|
||||
rolls_value = abs(_to_decimal(record.get('JianShu'), field_name='JianShu', default=Decimal('0')))
|
||||
zk_value = abs(_to_decimal(record.get('ZkJinE'), field_name='ZkJinE', default=Decimal('0')))
|
||||
total_amount += amount_value
|
||||
zk_amount += zk_value
|
||||
matched_product = _find_local_product_by_external_product_id(
|
||||
merchant=customer.merchant,
|
||||
external_product_id=str(record.get('HpID') or '').strip(),
|
||||
)
|
||||
product_name = getattr(matched_product, 'name', '') or str(record.get('HpID') or '').strip()
|
||||
unit = str(record.get('JiJiaDW') or '').strip() or getattr(matched_product, 'get_unit_display', lambda: '')()
|
||||
item_payload = {
|
||||
'product_id': getattr(matched_product, 'id', None),
|
||||
'product_name': product_name,
|
||||
'quantity': _decimal_to_string(quantity_value),
|
||||
'price': _decimal_to_string(price_value),
|
||||
'unit': unit,
|
||||
'color': '',
|
||||
'spec': getattr(matched_product, 'spec', '') or '',
|
||||
'quantity_of_rolls': [],
|
||||
'num_of_rolls': int(rolls_value) if rolls_value == rolls_value.to_integral_value() else 0,
|
||||
'external_sub_id': record.get('SubID'),
|
||||
'external_product_id': str(record.get('HpID') or '').strip(),
|
||||
}
|
||||
items_payload.append(item_payload)
|
||||
for note_key in ('BeiZhu', 'BeiZhuC', 'BeiZhuD', 'MeoD'):
|
||||
note = str(record.get(note_key) or '').strip()
|
||||
if note:
|
||||
remarks_lines.append(f'{note_key}: {note}')
|
||||
|
||||
return {
|
||||
'merchant': customer.merchant,
|
||||
'customer': customer,
|
||||
'category': category,
|
||||
'external_customer_id': external_customer_id,
|
||||
'external_source_id': external_source_id,
|
||||
'occurred_at': occurred_at,
|
||||
'recorded_at': recorded_at,
|
||||
'settlement_method': str(first_record.get('JieSunFS') or '').strip(),
|
||||
'total_amount': abs(total_amount),
|
||||
'sf_amount': sf_amount,
|
||||
'zk_amount': zk_amount,
|
||||
'remarks': '\n'.join(dict.fromkeys(remarks_lines)),
|
||||
'items_payload': items_payload,
|
||||
'extra_payload': {
|
||||
'raw_count': len(grouped_records),
|
||||
'dan_type': str(first_record.get('DanType') or '').strip(),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _is_equivalent_external_statement_order(
|
||||
*,
|
||||
existing: business_models.ExternalCustomerStatementOrder,
|
||||
normalized: dict[str, Any],
|
||||
) -> bool:
|
||||
return all(
|
||||
[
|
||||
existing.customer_id == normalized['customer'].id,
|
||||
existing.external_customer_id == normalized['external_customer_id'],
|
||||
existing.occurred_at == normalized['occurred_at'],
|
||||
existing.recorded_at == normalized['recorded_at'],
|
||||
existing.total_amount == normalized['total_amount'],
|
||||
existing.sf_amount == normalized.get('sf_amount', Decimal('0')),
|
||||
existing.zk_amount == normalized.get('zk_amount', Decimal('0')),
|
||||
existing.items_payload == normalized['items_payload'],
|
||||
(existing.remarks or '') == (normalized['remarks'] or ''),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
def _apply_sale_discount_zk(
|
||||
*,
|
||||
merchant: basic_models.Merchant,
|
||||
customer: basic_models.Customer,
|
||||
sale_discounts: list[dict[str, Any]],
|
||||
progress_callback: Callable[[str], None] | None = None,
|
||||
) -> int:
|
||||
"""
|
||||
从 sale_discount 数据（F_Skd 的 XS% 行）中提取 ZkJinE，
|
||||
更新到对应的 ExternalCustomerStatementOrder.zk_amount。
|
||||
返回更新的记录数。
|
||||
"""
|
||||
# 按 BianHaoID 聚合折扣（一个 BianHaoID 可能有多条 F_Skd 记录）
|
||||
zk_by_source: dict[str, Decimal] = {}
|
||||
for record in sale_discounts:
|
||||
if not isinstance(record, dict):
|
||||
continue
|
||||
source_id = str(record.get('BianHaoID') or '').strip()
|
||||
if not source_id:
|
||||
continue
|
||||
zk_value = abs(_to_decimal(record.get('ZkJinE'), field_name='ZkJinE', default=Decimal('0')))
|
||||
if zk_value:
|
||||
zk_by_source[source_id] = zk_by_source.get(source_id, Decimal('0')) + zk_value
|
||||
|
||||
if not zk_by_source:
|
||||
return 0
|
||||
|
||||
updated_count = 0
|
||||
for source_id, zk_total in zk_by_source.items():
|
||||
updated = business_models.ExternalCustomerStatementOrder.objects.filter(
|
||||
merchant=merchant,
|
||||
customer=customer,
|
||||
category=business_models.ExternalCustomerStatementCategoryEnum.SALE,
|
||||
external_source_id=source_id,
|
||||
).exclude(zk_amount=zk_total).update(zk_amount=zk_total)
|
||||
updated_count += updated
|
||||
|
||||
if updated_count:
|
||||
_emit_progress(
|
||||
progress_callback,
|
||||
f'销售折扣更新: {updated_count} 条记录的 zk_amount 已从 sale_discount 数据更新',
|
||||
)
|
||||
return updated_count
|
||||
|
||||
|
||||
def _find_local_product_by_external_product_id(
|
||||
*,
|
||||
merchant: basic_models.Merchant,
|
||||
external_product_id: str,
|
||||
):
|
||||
if not external_product_id:
|
||||
return None
|
||||
return basic_models.Product.objects.filter(merchant=merchant, human_id=external_product_id).first()
|
||||
|
||||
|
||||
def _parse_external_datetime(value: Any) -> datetime | None:
|
||||
text = str(value or '').strip()
|
||||
if not text:
|
||||
return None
|
||||
try:
|
||||
return datetime.fromisoformat(text.replace('Z', '+00:00'))
|
||||
except ValueError as exc:
|
||||
raise ExternalFinanceSyncError(f'外部日期时间格式非法: {text}') from exc
|
||||
|
||||
|
||||
def _decimal_to_string(value: Decimal) -> str:
|
||||
return f'{value:.2f}'
|
||||
|
||||
|
||||
def _emit_progress(progress_callback: Callable[[str], None] | None, message: str) -> None:
|
||||
if callable(progress_callback):
|
||||
progress_callback(message)
|
||||
Reference in New Issue
Block a user