"""市场专员/地区经理移动端的催缴与按样本缴费服务。""" from __future__ import annotations from datetime import date, datetime, timedelta from decimal import Decimal from typing import Any, Dict, List, Optional, Tuple from fastapi import HTTPException from sqlalchemy import Integer, and_, case, cast, exists, func, or_, select from sqlalchemy.orm import Session, aliased from app.core.utils import parse_datetime from app.crud import protocol_sample_payment as payment_crud from app.models.models import ( BizCellSampleType, BizCollectionProtocol, BizProtocolCollectionInfo, BizProtocolPartyCustomer, BizProtocolPaymentSchedule, BizProtocolPaymentTransfer, BizProtocolSample, BizProtocolSamplePayment, SysUser, ) from app.services.market_field_scope import ( MarketFieldContext, apply_cell_protocol_scope, ) from app.services.payment_chase import ( chase_period_money, is_chase_candidate, parse_due_date_ymd, ) from app.services.sample_payment_progress import ( compute_sample_payment_progress, is_schedule_effectively_cleared, ) from app.services.market_support import ( validate_attachment_refs, ) def _current_dept_id(user_info: dict) -> str: return str( user_info.get('dept_id') or user_info.get('deptId') or user_info.get('region_id') or '' ).strip() def _fmt_time(value: Any) -> str: return value.strftime('%Y-%m-%d %H:%M:%S') if value else '' def _money(value: Any) -> Optional[str]: """以固定两位小数字符串输出，避免 JSON float 破坏 DECIMAL(18,2)。""" if value is None or value == '': return None return format(Decimal(str(value)).quantize(Decimal('0.01')), 'f') def _display_ymd(value: Any) -> str: text = str(value or '').strip() if len(text) == 8 and text.isdigit(): return f'{text[:4]}-{text[4:6]}-{text[6:8]}' return text def _load_sample_meta( db: Session, payments: List[BizProtocolSamplePayment], tenant_no: str, ) -> Dict[str, Dict[str, Any]]: sample_codes = list({str(row.sample_code or '').strip() for row in payments if row.sample_code}) if not sample_codes: return {} rows = ( db.query(BizProtocolSample, BizCellSampleType) .outerjoin( BizCellSampleType, and_( BizCellSampleType.id == BizProtocolSample.sample_type_id, BizCellSampleType.tenant_no == tenant_no, BizCellSampleType.del_flag == 0, ), ) .filter( BizProtocolSample.sample_code.in_(sample_codes), BizProtocolSample.tenant_no == tenant_no, BizProtocolSample.del_flag == 0, ) .order_by( BizProtocolSample.update_time.desc(), BizProtocolSample.create_time.desc(), BizProtocolSample.id.desc(), ) .all() ) result: Dict[str, Dict[str, Any]] = {} for sample, sample_type in rows: original = sample.original_price current = sample.price discount = None if original is not None and current is not None and original > current: discount = original - current result.setdefault( sample.sample_code or '', { 'storageYears': sample.storage_years, 'sampleTypeName': sample_type.type_name if sample_type else '', 'discountAmount': _money(discount), }, ) return result def _transfer_dict(row: BizProtocolPaymentTransfer) -> Dict[str, Any]: return { 'id': row.id, 'scheduleId': row.schedule_id, 'samplePaymentId': row.sample_payment_id, 'collectionProtocolId': row.collection_protocol_id, 'sampleCode': row.sample_code or '', 'transferAmount': _money(row.transfer_amount), 'transferTime': _fmt_time(row.transfer_time), 'transferType': row.transfer_type or '', 'transferAttachments': row.transfer_attachments or '', 'payStatus': row.pay_status or '', 'remark': row.remark or '', 'financeRemark': row.finance_remark or '', 'createTime': _fmt_time(row.create_time), 'updateTime': _fmt_time(row.update_time), } def _period_dict( row: BizProtocolPaymentSchedule, transfers: List[BizProtocolPaymentTransfer], ) -> Dict[str, Any]: paid, arrears = chase_period_money( due_amount=row.due_amount, transfers=[ {'transferAmount': item.transfer_amount, 'payStatus': item.pay_status} for item in transfers ], ) schedule_status = row.schedule_status or '' if schedule_status != 'void' and is_schedule_effectively_cleared( {'scheduleStatus': schedule_status, 'dueAmount': row.due_amount}, confirmed_paid=Decimal(str(paid or 0)), ): schedule_status = 'cleared' return { 'id': row.id, 'scheduleNo': row.schedule_no, 'dueAmount': _money(row.due_amount), 'dueDate': row.due_date or '', 'scheduleStatus': schedule_status, 'paidAmount': _money(paid), 'arrearsAmount': _money(arrears), 'transfers': [_transfer_dict(item) for item in transfers], } def _payment_dict( payment: BizProtocolSamplePayment, protocol: BizCollectionProtocol, party: Optional[BizProtocolPartyCustomer], schedules: List[BizProtocolPaymentSchedule], transfers_by_schedule: Dict[str, List[BizProtocolPaymentTransfer]], sample_meta: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: meta = sample_meta or {} schedule_rows = [ _period_dict(row, transfers_by_schedule.get(row.id, [])) for row in schedules ] active_schedule_rows = [ schedule_row for schedule_row, schedule in zip(schedule_rows, schedules) if (schedule.schedule_status or '') != 'void' ] transfers = [ transfer for schedule in schedules for transfer in transfers_by_schedule.get(schedule.id, []) ] periods = payment_crud._build_periods_array(schedules, transfers) period_by_schedule: Dict[str, Dict[str, Any]] = {} for period in periods: if not period: continue schedule_id = str(period.get('scheduleId') or '') period['transfers'] = [ _transfer_dict(row) for row in transfers_by_schedule.get(schedule_id, []) ] period_by_schedule[schedule_id] = period for schedule in schedule_rows: period = period_by_schedule.get(str(schedule.get('id') or ''), {}) schedule['displayPaidAmount'] = _money(period.get('displayPaidAmount')) schedule['payDate'] = period.get('payDate') or '' progress_summary = compute_sample_payment_progress( schedules=[ { 'id': row.id, 'scheduleStatus': row.schedule_status or '', 'dueAmount': row.due_amount, } for row in schedules ], transfers=[ { 'scheduleId': row.schedule_id, 'payStatus': row.pay_status or '', 'transferAmount': row.transfer_amount, } for row in transfers ], ) paid_total = progress_summary['paidAmount'] remain_total = progress_summary['dueRemainAmount'] receivable_total = sum( ( Decimal(str(row.due_amount or '0')) for row in schedules if (row.schedule_status or '') != 'void' ), Decimal('0'), ) progress_code = progress_summary['progressStatus'] progress = { 'yjq': '已缴清', 'bfyj': '部分已缴', 'dfj': '待缴费', }[progress_code] return { 'id': payment.id, 'collectionProtocolId': protocol.id, 'protocolNo': protocol.protocol_no or '', 'protocolStatus': protocol.protocol_status or '', 'paymentModelVersion': int(getattr(protocol, 'payment_model_version', None) or 1), 'motherName': (party.mother_name if party else '') or '', 'motherPhone': (party.mother_phone if party else '') or '', 'sampleCode': payment.sample_code or '', 'payMode': payment.pay_mode or '', 'contractAmount': _money(payment.contract_amount), 'signDate': _display_ymd(getattr(protocol, 'register_date', None)), 'storageYears': meta.get('storageYears'), 'sampleTypeName': meta.get('sampleTypeName') or '', 'discountAmount': meta.get('discountAmount'), 'receivableAmount': _money(receivable_total), 'remark': payment.remark or '', 'paidAmount': _money(paid_total), 'remainAmount': _money(remain_total), 'dueRemainAmount': _money(remain_total), 'clearedScheduleCount': progress_summary['clearedScheduleCount'], 'activeScheduleCount': progress_summary['activeScheduleCount'], 'paymentProgress': progress, 'progressStatus': progress_code, 'periods': periods, 'schedules': schedule_rows, } def _scoped_payment_query( db: Session, *, ctx: MarketFieldContext, user_info: dict, v2_only: bool = False, ): protocol_conditions = [ BizCollectionProtocol.id == BizProtocolSamplePayment.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0, ] if v2_only: protocol_conditions.append(BizCollectionProtocol.payment_model_version == 2) query = ( db.query( BizProtocolSamplePayment, BizCollectionProtocol, BizProtocolPartyCustomer, ) .join( BizCollectionProtocol, and_(*protocol_conditions), ) .outerjoin( BizProtocolPartyCustomer, and_( BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id, BizProtocolPartyCustomer.tenant_no == ctx.tenant_no, BizProtocolPartyCustomer.del_flag == 0, ), ) .filter( BizProtocolSamplePayment.tenant_no == ctx.tenant_no, BizProtocolSamplePayment.del_flag == 0, ) ) return apply_cell_protocol_scope(query, db, ctx, user_info=user_info) def _load_children( db: Session, payment_ids: List[str], tenant_no: str, ) -> Tuple[ Dict[str, List[BizProtocolPaymentSchedule]], Dict[str, List[BizProtocolPaymentTransfer]], ]: if not payment_ids: return {}, {} schedules = ( db.query(BizProtocolPaymentSchedule) .filter( BizProtocolPaymentSchedule.sample_payment_id.in_(payment_ids), BizProtocolPaymentSchedule.tenant_no == tenant_no, BizProtocolPaymentSchedule.del_flag == 0, ) .order_by( BizProtocolPaymentSchedule.order_num.asc(), BizProtocolPaymentSchedule.schedule_no.asc(), ) .all() ) schedule_ids = [row.id for row in schedules] transfers = [] if schedule_ids: transfers = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.schedule_id.in_(schedule_ids), BizProtocolPaymentTransfer.tenant_no == tenant_no, BizProtocolPaymentTransfer.del_flag == 0, ) .order_by(BizProtocolPaymentTransfer.create_time.asc()) .all() ) schedules_by_payment: Dict[str, List[BizProtocolPaymentSchedule]] = {} transfers_by_schedule: Dict[str, List[BizProtocolPaymentTransfer]] = {} for row in schedules: schedules_by_payment.setdefault(row.sample_payment_id, []).append(row) for row in transfers: transfers_by_schedule.setdefault(row.schedule_id, []).append(row) return schedules_by_payment, transfers_by_schedule def list_payments( db: Session, *, ctx: MarketFieldContext, user_info: dict, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, pay_mode: Optional[str] = None, progress_status: Optional[str] = None, installment_due_date_start: Optional[str] = None, installment_due_date_end: Optional[str] = None, ) -> Tuple[List[Dict[str, Any]], int]: query = _scoped_payment_query(db, ctx=ctx, user_info=user_info) keyword_value = (keyword or '').strip() if keyword_value: like = f'%{keyword_value}%' query = query.filter( or_( BizCollectionProtocol.protocol_no.like(like), BizProtocolSamplePayment.sample_code.like(like), BizProtocolPartyCustomer.mother_name.like(like), BizProtocolPartyCustomer.mother_phone.like(like), ) ) mode = (pay_mode or '').strip() if mode: query = query.filter(BizProtocolSamplePayment.pay_mode == mode) def due_date_value(raw: Optional[str]) -> str: text = str(raw or '').strip().replace('-', '') if not text: return '' try: datetime.strptime(text, '%Y%m%d') except ValueError as exc: raise ValueError('应缴费时间格式无效，应为 YYYY-MM-DD') from exc return text due_start = due_date_value(installment_due_date_start) due_end = due_date_value(installment_due_date_end) if due_start or due_end: due_conditions = [ BizProtocolPaymentSchedule.sample_payment_id == BizProtocolSamplePayment.id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', ] normalized_due = func.replace(BizProtocolPaymentSchedule.due_date, '-', '') if due_start: due_conditions.append(normalized_due >= due_start) if due_end: due_conditions.append(normalized_due <= due_end) query = query.filter(exists().where(and_(*due_conditions))) progress_code = str(progress_status or '').strip() if progress_code and progress_code not in {'dfj', 'bfyj', 'yjq'}: raise ValueError('缴费进度无效') if progress_code: schedule_count = ( select(func.count(BizProtocolPaymentSchedule.id)) .where( BizProtocolPaymentSchedule.sample_payment_id == BizProtocolSamplePayment.id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', ) .correlate(BizProtocolSamplePayment) .scalar_subquery() ) schedule_confirmed_paid = ( select( func.coalesce( func.sum(BizProtocolPaymentTransfer.transfer_amount), 0, ) ) .where( BizProtocolPaymentTransfer.schedule_id == BizProtocolPaymentSchedule.id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'yjf', ) .correlate(BizProtocolPaymentSchedule) .scalar_subquery() ) # 未清期：非 void，且非显式 cleared，且 yjf 合计仍小于应缴 unsettled_exists = ( exists() .where( BizProtocolPaymentSchedule.sample_payment_id == BizProtocolSamplePayment.id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', BizProtocolPaymentSchedule.schedule_status != 'cleared', schedule_confirmed_paid < func.coalesce(BizProtocolPaymentSchedule.due_amount, 0), ) .correlate(BizProtocolSamplePayment) ) confirmed_paid = ( select(func.coalesce(func.sum(BizProtocolPaymentTransfer.transfer_amount), 0)) .select_from(BizProtocolPaymentTransfer) .join( BizProtocolPaymentSchedule, and_( BizProtocolPaymentSchedule.id == BizProtocolPaymentTransfer.schedule_id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, ), ) .where( BizProtocolPaymentSchedule.sample_payment_id == BizProtocolSamplePayment.id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'yjf', ) .correlate(BizProtocolSamplePayment) .scalar_subquery() ) is_cleared = and_( schedule_count > 0, ~unsettled_exists, ) if progress_code == 'yjq': query = query.filter(is_cleared) elif progress_code == 'bfyj': query = query.filter(confirmed_paid > 0, ~is_cleared) else: query = query.filter(confirmed_paid <= 0, ~is_cleared) total = ( query.order_by(None) .with_entities(func.count(BizProtocolSamplePayment.id)) .scalar() or 0 ) page = max(1, page) page_size = min(max(1, page_size), 100) rows = ( query.order_by( BizProtocolSamplePayment.create_time.desc(), BizProtocolSamplePayment.id.desc(), ) .offset((page - 1) * page_size) .limit(page_size) .all() ) schedules, transfers = _load_children( db, [row[0].id for row in rows], ctx.tenant_no ) sample_meta = _load_sample_meta(db, [row[0] for row in rows], ctx.tenant_no) return [ _payment_dict( payment, protocol, party, schedules.get(payment.id, []), transfers, sample_meta.get(payment.sample_code or ''), ) for payment, protocol, party in rows ], int(total) def get_payment_detail( db: Session, *, payment_id: str, ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, Any]: row = _scoped_payment_query( db, ctx=ctx, user_info=user_info, v2_only=False ).filter( BizProtocolSamplePayment.id == payment_id ).first() if not row: raise HTTPException(status_code=404, detail='缴费方案不存在或无权访问') payment, protocol, party = row schedules, transfers = _load_children(db, [payment.id], ctx.tenant_no) sample_meta = _load_sample_meta(db, [payment], ctx.tenant_no) return _payment_dict( payment, protocol, party, schedules.get(payment.id, []), transfers, sample_meta.get(payment.sample_code or ''), ) def list_payment_chase( db: Session, *, ctx: MarketFieldContext, user_info: dict, page: int = 1, page_size: int = 10, keyword: Optional[str] = None, chase_type: Optional[str] = None, pay_mode: Optional[str] = None, sales_user_id: Optional[str] = None, today: Optional[date] = None, ) -> Tuple[List[Dict[str, Any]], int]: today_value = today or date.today() today_key = today_value.strftime('%Y%m%d') end_key = (today_value + timedelta(days=30)).strftime('%Y%m%d') normalized_due = func.replace(BizProtocolPaymentSchedule.due_date, '-', '') dialect_name = db.get_bind().dialect.name if dialect_name == 'mysql': parsed_due = func.str_to_date(normalized_due, '%Y%m%d') month_start = func.str_to_date( func.concat(func.substr(normalized_due, 1, 6), '01'), '%Y%m%d', ) month_number = cast(func.substr(normalized_due, 5, 2), Integer) day_number = cast(func.substr(normalized_due, 7, 2), Integer) valid_due = and_( func.length(normalized_due) == 8, normalized_due.op('REGEXP')('^[0-9]{8}$'), month_number.between(1, 12), day_number.between(1, func.day(func.last_day(month_start))), func.date_format(parsed_due, '%Y%m%d') == normalized_due, ) else: parsed_due = func.date( func.printf( '%s-%s-%s', func.substr(normalized_due, 1, 4), func.substr(normalized_due, 5, 2), func.substr(normalized_due, 7, 2), ) ) valid_due = and_( func.length(normalized_due) == 8, func.strftime('%Y%m%d', parsed_due) == normalized_due, ) paid_subquery = ( db.query( BizProtocolPaymentTransfer.schedule_id.label('schedule_id'), func.sum( case( ( BizProtocolPaymentTransfer.pay_status == 'yjf', BizProtocolPaymentTransfer.transfer_amount, ), else_=Decimal('0'), ) ).label('paid_amount'), ) .filter( BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, ) .group_by(BizProtocolPaymentTransfer.schedule_id) .subquery() ) paid_amount = func.coalesce(paid_subquery.c.paid_amount, Decimal('0')) arrears_amount = ( func.coalesce(BizProtocolPaymentSchedule.due_amount, Decimal('0')) - paid_amount ) collection_row = aliased(BizProtocolCollectionInfo) query = ( db.query( BizProtocolPaymentSchedule, BizProtocolSamplePayment, BizCollectionProtocol, BizProtocolPartyCustomer, collection_row, paid_amount.label('paid_amount'), ) .join( BizProtocolSamplePayment, and_( BizProtocolSamplePayment.id == BizProtocolPaymentSchedule.sample_payment_id, BizProtocolSamplePayment.collection_protocol_id == BizProtocolPaymentSchedule.collection_protocol_id, BizProtocolSamplePayment.tenant_no == ctx.tenant_no, BizProtocolSamplePayment.del_flag == 0, ), ) .join( BizCollectionProtocol, and_( BizCollectionProtocol.id == BizProtocolPaymentSchedule.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0, ), ) .join( collection_row, and_( collection_row.collection_protocol_id == BizCollectionProtocol.id, collection_row.tenant_no == ctx.tenant_no, collection_row.del_flag == 0, ), ) .outerjoin( BizProtocolPartyCustomer, and_( BizProtocolPartyCustomer.collection_protocol_id == BizCollectionProtocol.id, BizProtocolPartyCustomer.tenant_no == ctx.tenant_no, BizProtocolPartyCustomer.del_flag == 0, ), ) .outerjoin( paid_subquery, paid_subquery.c.schedule_id == BizProtocolPaymentSchedule.id, ) .filter( BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', valid_due, normalized_due <= end_key, arrears_amount > 0, ) ) query = apply_cell_protocol_scope(query, db, ctx, user_info=user_info) keyword_value = (keyword or '').strip() if keyword_value: like = f'%{keyword_value}%' query = query.filter( or_( BizCollectionProtocol.protocol_no.like(like), BizProtocolSamplePayment.sample_code.like(like), BizProtocolPartyCustomer.mother_name.like(like), BizProtocolPartyCustomer.mother_phone.like(like), ) ) mode = (pay_mode or '').strip() if mode: query = query.filter(BizProtocolSamplePayment.pay_mode == mode) sales = (sales_user_id or '').strip() if sales: query = query.filter(collection_row.sales_user_id == sales) kind = (chase_type or '').strip() if kind == 'overdue': query = query.filter(normalized_due < today_key) elif kind == 'upcoming': query = query.filter(normalized_due >= today_key) total = ( query.order_by(None) .with_entities(func.count(BizProtocolPaymentSchedule.id)) .scalar() or 0 ) page = max(1, page) page_size = min(max(1, page_size), 100) candidates = ( query.order_by( normalized_due.asc(), BizCollectionProtocol.protocol_no.asc(), BizProtocolPaymentSchedule.id.asc(), ) .offset((page - 1) * page_size) .limit(page_size) .all() ) sales_user_ids = { collection.sales_user_id for _, _, _, _, collection, _ in candidates if collection.sales_user_id } sales_user_names = { row.id: row.nick_name or '' for row in db.query(SysUser).filter( SysUser.id.in_(sales_user_ids or ['']), SysUser.tenant_no == ctx.tenant_no, ) } rows: List[Dict[str, Any]] = [] for schedule, payment, protocol, party, collection, paid in candidates: due_date = parse_due_date_ymd(schedule.due_date) if due_date is None: continue paid_value = Decimal(str(paid or 0)) due_value = Decimal(str(schedule.due_amount or 0)) arrears = max(due_value - paid_value, Decimal('0')) is_overdue = due_date < today_value rows.append( { 'scheduleId': schedule.id, 'samplePaymentId': payment.id, 'protocolId': protocol.id, 'protocolNo': protocol.protocol_no or '', 'protocolStatus': protocol.protocol_status or '', 'motherName': (party.mother_name if party else '') or '', 'motherPhone': (party.mother_phone if party else '') or '', 'sampleCode': payment.sample_code or schedule.sample_code or '', 'salesUserId': collection.sales_user_id or '', 'salesUserName': sales_user_names.get( collection.sales_user_id or '', '' ), 'payMode': payment.pay_mode or '', 'scheduleNo': schedule.schedule_no, 'dueDate': due_date.strftime('%Y-%m-%d'), 'dueAmount': _money(schedule.due_amount), 'paidAmount': _money(paid_value), 'arrearsAmount': _money(arrears), 'chaseStatus': 'overdue' if is_overdue else 'upcoming', 'daysOverdue': (today_value - due_date).days if is_overdue else 0, 'daysUntilDue': (due_date - today_value).days if not is_overdue else 0, 'paymentModelVersion': int(protocol.payment_model_version or 1), } ) return rows, int(total) def _scoped_chase_schedule( db: Session, *, schedule_id: str, ctx: MarketFieldContext, user_info: dict, today: Optional[date] = None, ): row = ( apply_cell_protocol_scope( db.query( BizProtocolPaymentSchedule, BizProtocolSamplePayment, BizCollectionProtocol, ) .join( BizProtocolSamplePayment, and_( BizProtocolSamplePayment.id == BizProtocolPaymentSchedule.sample_payment_id, BizProtocolSamplePayment.collection_protocol_id == BizProtocolPaymentSchedule.collection_protocol_id, BizProtocolSamplePayment.tenant_no == ctx.tenant_no, BizProtocolSamplePayment.del_flag == 0, ), ) .join( BizCollectionProtocol, and_( BizCollectionProtocol.id == BizProtocolPaymentSchedule.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0, ), ) .filter( BizProtocolPaymentSchedule.id == schedule_id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', ), db, ctx, user_info=user_info, ) .first() ) if not row: raise HTTPException(status_code=404, detail='催缴期次不存在或无权访问') schedule, payment, protocol = row due_date = parse_due_date_ymd(schedule.due_date) confirmed_paid = ( db.query(func.coalesce(func.sum(BizProtocolPaymentTransfer.transfer_amount), 0)) .filter( BizProtocolPaymentTransfer.schedule_id == schedule.id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'yjf', ) .scalar() or Decimal('0') ) arrears = max( Decimal(str(schedule.due_amount or 0)) - Decimal(str(confirmed_paid or 0)), Decimal('0'), ) if due_date is None or not is_chase_candidate( due_date=due_date, today=today or date.today(), arrears=arrears, ): raise HTTPException(status_code=404, detail='催缴期次不存在或无权访问') return schedule, payment, protocol def get_payment_chase_detail( db: Session, *, schedule_id: str, view: str, ctx: MarketFieldContext, user_info: dict, today: Optional[date] = None, ) -> Dict[str, Any]: schedule, payment, protocol = _scoped_chase_schedule( db, schedule_id=schedule_id, ctx=ctx, user_info=user_info, today=today, ) target_view = str(view or '').strip() if target_view == 'payment': detail = get_payment_detail( db, payment_id=payment.id, ctx=ctx, user_info=user_info, ) target_transfers = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.schedule_id == schedule.id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, ) .order_by(BizProtocolPaymentTransfer.create_time.asc()) .all() ) target_detail = _payment_dict( payment, protocol, None, [schedule], {schedule.id: target_transfers}, ) for key in ( 'receivableAmount', 'paidAmount', 'remainAmount', 'dueRemainAmount', 'clearedScheduleCount', 'activeScheduleCount', 'paymentProgress', 'progressStatus', 'periods', 'schedules', ): detail[key] = target_detail[key] elif target_view == 'protocol': from app.services import market_protocol detail = market_protocol.get_protocol( db, source=market_protocol.SOURCE_CELL, protocol_id=protocol.id, ctx=ctx, user_info=user_info, ) else: raise ValueError('催缴详情类型无效') detail['chaseView'] = target_view detail['targetScheduleId'] = schedule.id detail['allowedActions'] = [] return detail def _scoped_schedule( db: Session, *, schedule_id: str, ctx: MarketFieldContext, user_info: dict, ): query = ( db.query(BizProtocolPaymentSchedule) .join( BizProtocolSamplePayment, and_( BizProtocolSamplePayment.id == BizProtocolPaymentSchedule.sample_payment_id, BizProtocolSamplePayment.collection_protocol_id == BizProtocolPaymentSchedule.collection_protocol_id, BizProtocolSamplePayment.tenant_no == ctx.tenant_no, BizProtocolSamplePayment.del_flag == 0, ), ) .join( BizCollectionProtocol, and_( BizCollectionProtocol.id == BizProtocolPaymentSchedule.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0, ), ) .filter( BizProtocolPaymentSchedule.id == schedule_id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, BizProtocolPaymentSchedule.schedule_status != 'void', ) ) query = apply_cell_protocol_scope(query, db, ctx, user_info=user_info) row = query.populate_existing().with_for_update().first() if not row: raise HTTPException(status_code=404, detail='应缴批次不存在或无权访问') return row def _scoped_transfer( db: Session, *, transfer_id: str, ctx: MarketFieldContext, user_info: dict, ): parent = ( db.query( BizProtocolPaymentTransfer.schedule_id, BizProtocolPaymentTransfer.sample_payment_id, BizProtocolPaymentTransfer.collection_protocol_id, ) .filter( BizProtocolPaymentTransfer.id == transfer_id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, ) .first() ) if not parent: raise HTTPException(status_code=404, detail='转账记录不存在或无权访问') schedule_query = ( db.query(BizProtocolPaymentSchedule) .join( BizProtocolSamplePayment, and_( BizProtocolSamplePayment.id == parent.sample_payment_id, BizProtocolSamplePayment.id == BizProtocolPaymentSchedule.sample_payment_id, BizProtocolSamplePayment.collection_protocol_id == parent.collection_protocol_id, BizProtocolSamplePayment.tenant_no == ctx.tenant_no, BizProtocolSamplePayment.del_flag == 0, ), ) .join( BizCollectionProtocol, and_( BizCollectionProtocol.id == parent.collection_protocol_id, BizCollectionProtocol.id == BizProtocolPaymentSchedule.collection_protocol_id, BizCollectionProtocol.tenant_no == ctx.tenant_no, BizCollectionProtocol.del_flag == 0, ), ) .filter( BizProtocolPaymentSchedule.id == parent.schedule_id, BizProtocolPaymentSchedule.sample_payment_id == parent.sample_payment_id, BizProtocolPaymentSchedule.collection_protocol_id == parent.collection_protocol_id, BizProtocolPaymentSchedule.tenant_no == ctx.tenant_no, BizProtocolPaymentSchedule.del_flag == 0, ) ) schedule_query = apply_cell_protocol_scope( schedule_query, db, ctx, user_info=user_info ) schedule = schedule_query.populate_existing().with_for_update().first() if not schedule: raise HTTPException(status_code=404, detail='转账记录不存在或无权访问') row = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.id == transfer_id, BizProtocolPaymentTransfer.schedule_id == schedule.id, BizProtocolPaymentTransfer.sample_payment_id == parent.sample_payment_id, BizProtocolPaymentTransfer.collection_protocol_id == parent.collection_protocol_id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, ) .populate_existing() .with_for_update() .first() ) if not row: raise HTTPException(status_code=404, detail='转账记录不存在或无权访问') return row def _editable_fields(fields: Dict[str, Any]) -> Dict[str, Any]: aliases = { 'transfer_amount': 'transferAmount', 'transfer_time': 'transferTime', 'transfer_type': 'transferType', 'transfer_attachments': 'transferAttachments', 'remark': 'remark', } result: Dict[str, Any] = {} for python_name, alias in aliases.items(): if python_name in fields: result[alias] = fields[python_name] elif alias in fields: result[alias] = fields[alias] amount = result.get('transferAmount') if amount is not None and Decimal(str(amount)) <= 0: raise ValueError('转账金额必须大于 0') if 'transferTime' in result and result['transferTime'] not in (None, ''): value = result['transferTime'] result['transferTime'] = value if isinstance(value, datetime) else parse_datetime(str(value)) if 'transferType' in result: result['transferType'] = str(result['transferType'] or '').strip() or None if 'transferAttachments' in result: result['transferAttachments'] = str( result['transferAttachments'] or '' ).strip() or None if 'remark' in result: result['remark'] = str(result['remark'] or '').strip() or None return result def create_transfer( db: Session, *, schedule_id: str, fields: Dict[str, Any], ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, Any]: try: schedule = _scoped_schedule( db, schedule_id=schedule_id, ctx=ctx, user_info=user_info, ) confirmed_paid = ( db.query( func.coalesce(func.sum(BizProtocolPaymentTransfer.transfer_amount), 0) ) .filter( BizProtocolPaymentTransfer.schedule_id == schedule.id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'yjf', ) .scalar() or Decimal('0') ) arrears = max( Decimal(str(schedule.due_amount or 0)) - Decimal(str(confirmed_paid or 0)), Decimal('0'), ) if arrears <= 0: raise ValueError('当前应缴批次无欠缴金额') values = _editable_fields(fields) row = payment_crud.add_transfer( db, schedule_id=schedule_id, tenant_no=ctx.tenant_no, user_id=ctx.user_id, transfer_amount=values.get('transferAmount'), transfer_time=values.get('transferTime'), transfer_type=values.get('transferType'), transfer_attachments=values.get('transferAttachments'), remark=values.get('remark'), create_dept_id=_current_dept_id(user_info), commit=False, ) if 'transferAttachments' in values: row.transfer_attachments = validate_attachment_refs( db, values['transferAttachments'], module='payment', ctx=ctx, creating=True, business_id=row.id, ) db.add(row) db.flush() result = _transfer_dict(row) db.commit() return result except Exception: db.rollback() raise def update_transfer( db: Session, *, transfer_id: str, fields: Dict[str, Any], ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, Any]: try: row = _scoped_transfer( db, transfer_id=transfer_id, ctx=ctx, user_info=user_info ) if (row.pay_status or '') != 'dfj': raise ValueError('仅已付款状态可编辑') values = _editable_fields(fields) if 'transferAttachments' in values: values['transferAttachments'] = validate_attachment_refs( db, values['transferAttachments'], module='payment', ctx=ctx, creating=False, existing_refs=row.transfer_attachments, business_id=row.id, ) if 'transferAmount' in values: row.transfer_amount = Decimal(str(values['transferAmount'])) if values['transferAmount'] is not None else None if 'transferTime' in values: row.transfer_time = values['transferTime'] if 'transferType' in values: row.transfer_type = values['transferType'] if 'transferAttachments' in values: row.transfer_attachments = values['transferAttachments'] if 'remark' in values: row.remark = values['remark'] row.update_by = ctx.user_id row.update_time = datetime.now() db.add(row) db.commit() db.refresh(row) return _transfer_dict(row) except Exception: db.rollback() raise def submit_transfer( db: Session, *, transfer_id: str, ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, Any]: try: row = _scoped_transfer( db, transfer_id=transfer_id, ctx=ctx, user_info=user_info ) if (row.pay_status or '') != 'dfj': raise ValueError('当前状态不可提交财务确认') if row.transfer_amount is None or Decimal(str(row.transfer_amount)) <= 0: raise ValueError('请填写转账金额') if not (row.transfer_type or '').strip(): raise ValueError('请选择转账方式') now = datetime.now() changed = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.id == transfer_id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'dfj', ) .update( { BizProtocolPaymentTransfer.pay_status: 'jfqrz', BizProtocolPaymentTransfer.update_by: ctx.user_id, BizProtocolPaymentTransfer.update_time: now, }, synchronize_session=False, ) ) if changed != 1: raise ValueError('转账状态已变化，请刷新后重试') db.commit() db.expire(row) return _transfer_dict(row) except Exception: db.rollback() raise def delete_transfer( db: Session, *, transfer_id: str, ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, str]: try: row = _scoped_transfer( db, transfer_id=transfer_id, ctx=ctx, user_info=user_info ) if (row.pay_status or '') != 'dfj': raise ValueError('当前状态不可删除转账记录') changed = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.id == transfer_id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'dfj', ) .update( { BizProtocolPaymentTransfer.del_flag: 1, BizProtocolPaymentTransfer.update_by: ctx.user_id, BizProtocolPaymentTransfer.update_time: datetime.now(), }, synchronize_session=False, ) ) if changed != 1: raise ValueError('转账状态已变化，请刷新后重试') db.commit() return {'id': transfer_id} except Exception: db.rollback() raise def withdraw_transfer( db: Session, *, transfer_id: str, ctx: MarketFieldContext, user_info: dict, ) -> Dict[str, Any]: try: row = _scoped_transfer( db, transfer_id=transfer_id, ctx=ctx, user_info=user_info ) if (row.pay_status or '') != 'jfqrz': raise ValueError('仅付款确认中可撤回') now = datetime.now() changed = ( db.query(BizProtocolPaymentTransfer) .filter( BizProtocolPaymentTransfer.id == transfer_id, BizProtocolPaymentTransfer.tenant_no == ctx.tenant_no, BizProtocolPaymentTransfer.del_flag == 0, BizProtocolPaymentTransfer.pay_status == 'jfqrz', ) .update( { BizProtocolPaymentTransfer.pay_status: 'dfj', BizProtocolPaymentTransfer.update_by: ctx.user_id, BizProtocolPaymentTransfer.update_time: now, }, synchronize_session=False, ) ) if changed != 1: raise ValueError('转账状态已变化，请刷新后重试') db.commit() db.expire(row) return _transfer_dict(row) except Exception: db.rollback() raise