import json from datetime import datetime, timezone, timedelta, date from sqlalchemy import func from app.extensions import db from app.models.purchase import PurchaseRequest from app.models.base import MaterialBase class PurchaseService: @staticmethod def generate_request_no(batch_seq: int = None): """ 生成采购单号: PUR-yyyyMMdd-HHmm-批次-批内序号 Args: batch_seq: 今日第几次采购批次（前端提交一批时传入，同一批共享） None 时回退为旧格式 PUR-日期-时间-流水 返回: - 有 batch_seq: PUR-20260831-1021-0001-0001（批次号-批内序号） - 无 batch_seq: PUR-20260831-1021-0001（旧格式兼容） """ beijing_tz = timezone(timedelta(hours=8)) now = datetime.now(beijing_tz) date_str = now.strftime('%Y%m%d') time_str = now.strftime('%H%M') if batch_seq is not None: # 批次前缀: PUR-日期-时间-批次 batch_prefix = f"PUR-{date_str}-{time_str}-{batch_seq:04d}" # 批内序号: 该批次前缀下的记录数 + 1 item_count = db.session.query(func.count(func.distinct(PurchaseRequest.request_no))) \ .filter(PurchaseRequest.request_no.like(f"{batch_prefix}-%")).scalar() return f"{batch_prefix}-{(item_count + 1):04d}" # 旧格式兼容: PUR-日期-时间-流水 prefix = f"PUR-{date_str}-{time_str}-" existing_count = db.session.query(func.count(func.distinct(PurchaseRequest.request_no))) \ .filter(PurchaseRequest.request_no.like(f"{prefix}%")).scalar() return f"{prefix}{(existing_count + 1):04d}" @staticmethod def get_next_batch_seq(): """ 返回今日下一个采购批次号（今日第几次） 统计今日已有的不同批次（单号第4段）数量，+1 """ beijing_tz = timezone(timedelta(hours=8)) now = datetime.now(beijing_tz) date_str = now.strftime('%Y%m%d') time_str = now.strftime('%H%M') prefix = f"PUR-{date_str}-{time_str}-" # 查询今日所有单号，提取第4段（批次号）去重 rows = db.session.query(PurchaseRequest.request_no) \ .filter(PurchaseRequest.request_no.like(f"{prefix}%")).all() batch_seqs = set() for (rn,) in rows: parts = rn.split('-') if len(parts) >= 4: batch_seqs.add(parts[3]) return len(batch_seqs) + 1 @staticmethod def auto_fill_from_material(keyword: str): """ 根据 name 或 spec_model 自动补全另一个字段 keyword: 用户输入的名称或规格 返回: {'name': ..., 'spec_model': ...} 或 None """ if not keyword: return None material = MaterialBase.query.filter( (MaterialBase.name.ilike(f'%{keyword}%')) | (MaterialBase.spec_model.ilike(f'%{keyword}%')) ).first() if material: return { 'name': material.name, 'spec_model': material.spec_model or '' } return None @staticmethod def create_purchase_request(data: dict, requester_id: int): """ 创建采购申请 data 包含: name, spec_model, quantity, purchase_date, supplier_link, remark, images, unit_price, total_price, approver_id, base_id (可选) """ batch_seq = data.get('batch_seq') request_no = PurchaseService.generate_request_no(batch_seq=batch_seq) purchase_date = data.get('purchase_date') if isinstance(purchase_date, str): purchase_date = datetime.strptime(purchase_date, '%Y-%m-%d').date() elif isinstance(purchase_date, datetime): purchase_date = purchase_date.date() # [新增] 自动匹配/关联基础物料 base_id = data.get('base_id') if not base_id and data.get('name'): # 尝试通过 name + spec_model 精确匹配 MaterialBase material = MaterialBase.query.filter( MaterialBase.name == data['name'], MaterialBase.spec_model == data.get('spec_model', ''), MaterialBase.is_enabled == True ).first() if material: base_id = material.id purchase = PurchaseRequest( request_no=request_no, base_id=base_id, # [新增] name=data['name'], spec_model=data.get('spec_model', ''), quantity=float(data['quantity']), purchase_date=purchase_date, supplier_link=data.get('supplier_link', ''), remark=data.get('remark', ''), images=json.dumps(data.get('images', []), ensure_ascii=False) if data.get('images') else '[]', unit_price=float(data.get('unit_price', 0) or 0), total_price=float(data.get('total_price', 0) or 0), tax_rate=float(data.get('tax_rate', 0) or 0), requester_id=requester_id, approver_id=data.get('approver_id'), status=0 ) db.session.add(purchase) db.session.commit() # 发送邮件给审批人 PurchaseService._notify_new_request(purchase) return purchase @staticmethod def approve_purchase_request(purchase_id: int, user_id: int, action: str, reject_reason: str = None): """ 审批采购申请 action: 'approve' 或 'reject' """ purchase = db.session.get(PurchaseRequest, purchase_id) if not purchase: raise ValueError("采购申请不存在") beijing_tz = timezone(timedelta(hours=8)) now = datetime.now(beijing_tz) # ★ 完结：库管将「已通过(1)」的申请单置为「已完结(4)」（仿出库审批） if action == 'close': if purchase.status != 1: raise ValueError("仅「已通过」的采购申请可完结") purchase.status = 4 purchase.approver_id = user_id purchase.approved_at = now db.session.commit() return purchase if purchase.status != 0: raise ValueError("当前状态不允许审批") if action == 'approve': purchase.status = 1 purchase.approver_id = user_id purchase.approved_at = now db.session.commit() PurchaseService._notify_approved(purchase) elif action == 'reject': purchase.status = 2 purchase.approver_id = user_id purchase.approved_at = now purchase.reject_reason = reject_reason or '' db.session.commit() PurchaseService._notify_rejected(purchase) else: raise ValueError("无效的审批操作") return purchase @staticmethod def get_purchase_list(page=1, per_page=20, requester_id=None, status=None): """获取采购申请列表，普通用户只看自己的，主管/超管看同公司全部""" from app.utils.decorators import get_current_company_filter from app.models.system import SysUser from sqlalchemy import or_ query = PurchaseRequest.query if requester_id is not None: query = query.filter(PurchaseRequest.requester_id == requester_id) if status is not None: query = query.filter(PurchaseRequest.status == status) # 【行级数据隔离】同公司可见：匹配 MaterialBase.company_name 或 SysUser.department company_limit = get_current_company_filter() if company_limit is not None: query = query.outerjoin(MaterialBase, PurchaseRequest.base_id == MaterialBase.id) \ .outerjoin(SysUser, PurchaseRequest.requester_id == SysUser.id) \ .filter(or_( MaterialBase.company_name == company_limit, SysUser.department == company_limit )) query = query.order_by(PurchaseRequest.created_at.desc()) pagination = query.paginate(page=page, per_page=per_page, error_out=False) return { 'items': [p.to_dict() for p in pagination.items], 'total': pagination.total, 'pages': pagination.pages, 'current_page': page } @staticmethod def get_purchase_by_id(purchase_id: int): purchase = db.session.get(PurchaseRequest, purchase_id) return purchase.to_dict() if purchase else None @staticmethod def get_approved_requests(page=1, per_page=20, keyword=None): """ 获取已审批通过且未入库的采购申请列表（专供库管按单入库使用） 筛选条件: - status == 1（已审批通过） - 尚未被任何 StockBuy 关联（request_id 未被引用） 返回字段包含: 采购申请信息 + MaterialBase 基础物料信息 """ from app.models.inbound.buy import StockBuy # 子查询：所有已被入库引用的 request_id（去重） stocked_ids = db.session.query(StockBuy.request_id).filter( StockBuy.request_id.isnot(None) ).distinct().subquery() # 主查询：已通过 且 不在已入库集合中 query = db.session.query(PurchaseRequest).filter( PurchaseRequest.status == 1 ).filter( ~PurchaseRequest.id.in_(stocked_ids) ) # 可选关键词搜索：采购单号 / 名称 / 规格 if keyword: k = f'%{keyword.strip()}%' query = query.filter( PurchaseRequest.request_no.ilike(k) | PurchaseRequest.name.ilike(k) | PurchaseRequest.spec_model.ilike(k) ) query = query.order_by(PurchaseRequest.approved_at.desc().nullslast(), PurchaseRequest.created_at.desc()) pagination = query.paginate(page=page, per_page=per_page, error_out=False) # ★ 批量预取 MaterialBase：三段式查询（base_id 精确 → name 精确 → name 模糊） purchase_items = list(pagination.items) base_ids = {p.base_id for p in purchase_items if p.base_id} names_no_base = [p.name for p in purchase_items if p.name and not p.base_id] material_map = {} # base_id → MaterialBase name_map = {} # (name, spec) or (name, '__fallback__') → MaterialBase # 阶段1: base_id 精确预取 if base_ids: materials = MaterialBase.query.filter(MaterialBase.id.in_(base_ids)).all() material_map = {m.id: m for m in materials} # 阶段2: name 精确预取（name.in_） if names_no_base: exact_matches = MaterialBase.query.filter( MaterialBase.name.in_(names_no_base), MaterialBase.is_enabled == True ).all() for m in exact_matches: name_map[(m.name, m.spec_model or '')] = m # 阶段3: name 模糊预取（仅>=2字符 + LIMIT 500 防爆炸） unmatched_names = [] for p in purchase_items: if p.base_id and p.base_id in material_map: continue if p.name and (p.name, p.spec_model or '') in name_map: continue n = (p.name or '').strip() if len(n) >= 2: # ★ 安全阈值：至少2字符才做模糊匹配 unmatched_names.append(n) if unmatched_names: fuzzy_conditions = [ MaterialBase.name.ilike(f'%{n}%') for n in unmatched_names ] fuzzy_matches = MaterialBase.query.filter( db.or_(*fuzzy_conditions), MaterialBase.is_enabled == True ).order_by(MaterialBase.id.desc()).limit(500).all() # ★ 硬上限：最多500条 for m in fuzzy_matches: key = (m.name, '__fuzzy__') if key not in name_map: name_map[key] = m # 内存匹配（O(1) 字典查找） items = [] for p in purchase_items: item = p.to_dict() material = None # 1. base_id 精确 if p.base_id: material = material_map.get(p.base_id) # 2. name + spec 精确 if not material and p.name: material = name_map.get((p.name, p.spec_model or '')) # 3. name 模糊回退 if not material and p.name: material = name_map.get((p.name, '__fuzzy__')) if material: item['material'] = { 'id': material.id, 'company_name': material.company_name or '', 'name': material.name, 'spec_model': material.spec_model or '', 'category': material.category or '', 'unit': material.unit or '', 'type': material.material_type or '', 'is_inspection_required': bool(material.is_inspection_required), } else: item['material'] = None items.append(item) return { 'items': items, 'total': pagination.total, 'pages': pagination.pages, 'current_page': page } @staticmethod def search_base_material(keyword: str, page: int = 1, limit: int = 20): """ 物料基础信息搜索，支持 name/spec_model/company_name 模糊匹配，返回分页结果 用于采购申请弹窗的物料远程搜索 """ from sqlalchemy import and_, or_ query = MaterialBase.query.filter(MaterialBase.is_enabled == True) if keyword: k = keyword.strip() k_str = f'%{k}%' query = query.filter(or_( MaterialBase.name.ilike(k_str), MaterialBase.spec_model.ilike(k_str), MaterialBase.company_name.ilike(k_str) )) query = query.order_by(MaterialBase.id.desc()) pagination = query.paginate(page=page, per_page=limit, error_out=False) items = [] for item in pagination.items: items.append({ 'id': item.id, 'company_name': item.company_name, 'name': item.name, 'spec_model': item.spec_model, 'category': item.category, 'unit': item.unit, 'type': item.material_type, 'pinyin': getattr(item, 'pinyin', ''), 'status': '启用' }) return { 'items': items, 'total': pagination.total, 'page': page, 'has_next': pagination.has_next } @staticmethod def _notify_new_request(purchase): """发送新申请邮件给审批人""" try: from app.utils.email_service import send_email_async from app.models.system import SysUser if not purchase.approver_id: return approver = db.session.get(SysUser, purchase.approver_id) if not approver or not approver.email: return subject = f"【待审批】采购申请单 {purchase.request_no}" content = f"""您好， 您有一笔新的采购申请待审批： 申请单号：{purchase.request_no} 采购物品：{purchase.name} 规格型号：{purchase.spec_model or '-'} 采购数量：{float(purchase.quantity)} 申请时间：{purchase.created_at.strftime('%Y-%m-%d %H:%M') if purchase.created_at else '-'} 备注说明：{purchase.remark or '无'} 请登录仓库管理系统进行审批。 此邮件由系统自动发送，请勿回复。 """ send_email_async(approver.email, subject, content) except Exception as e: try: from flask import current_app current_app.logger.error(f"[Email] 采购申请通知审批人失败: {e}") except Exception: print(f"[Email] 采购申请通知审批人失败: {e}") @staticmethod def _notify_approved(purchase): """审批通过后通知申请人""" try: from app.utils.email_service import send_email_async from app.models.system import SysUser requester = db.session.get(SysUser, purchase.requester_id) if not requester or not requester.email: return subject = f"【已通过】采购申请单 {purchase.request_no}" content = f"""{"尊敬的 " + requester.username + "，您好" if requester.username else "您好"}， 您的采购申请单 {purchase.request_no}（{purchase.name}）已审批通过，现已交给库管。 待库管完成入库后，您可在系统中查询采购记录。 此邮件由系统自动发送，请勿回复。 """ send_email_async(requester.email, subject, content) except Exception as e: try: from flask import current_app current_app.logger.error(f"[Email] 采购申请通过通知申请人失败: {e}") except Exception: print(f"[Email] 采购申请通过通知申请人失败: {e}") @staticmethod def _notify_rejected(purchase): """审批驳回后通知申请人""" try: from app.utils.email_service import send_email_async from app.models.system import SysUser requester = db.session.get(SysUser, purchase.requester_id) if not requester or not requester.email: return subject = f"【已驳回】采购申请单 {purchase.request_no}" content = f"""{"尊敬的 " + requester.username + "，您好" if requester.username else "您好"}， 您的采购申请单 {purchase.request_no}（{purchase.name}）已被驳回。 驳回原因：{purchase.reject_reason or '未说明'} 请登录仓库管理系统查看详情。 此邮件由系统自动发送，请勿回复。 """ send_email_async(requester.email, subject, content) except Exception as e: print(f"[Email] 采购申请驳回通知失败: {e}")