260 lines
11 KiB
Python
260 lines
11 KiB
Python
"""手工派车 CRUD：租户、角色数据范围和关联主数据校验。"""
|
|
import re
|
|
from datetime import date
|
|
from functools import wraps
|
|
from typing import Optional
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.exc import OperationalError
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.db_auth import DbUserContext
|
|
from app.core.utils import generate_uuid
|
|
from app.models.crm_models import CrmProduct
|
|
from app.models.pl_dispatch_calendar import PlDispatchCalendar
|
|
from app.models.pl_models import PlVehicle
|
|
from app.schemas.pl.dispatch_calendar import CreateBody, UpdateBody
|
|
from app.services.pl_data_scope import apply_pl_scope
|
|
|
|
CALENDAR_KINDS = ("acid", "alkali", "other", "site")
|
|
|
|
|
|
def _calendar_kind(kind: Optional[str]) -> str:
|
|
value = kind or "acid"
|
|
if value not in CALENDAR_KINDS:
|
|
raise HTTPException(status_code=422, detail="日历类型须为 acid、alkali、other 或 site")
|
|
return value
|
|
|
|
|
|
def _row_to_dict(row: PlDispatchCalendar) -> dict:
|
|
return {
|
|
"id": row.id,
|
|
"dispatchDate": row.dispatch_date,
|
|
"vehicleId": row.vehicle_id,
|
|
"licensePlate": row.license_plate,
|
|
"productId": row.product_id,
|
|
"productName": row.product_name,
|
|
"origin": row.origin,
|
|
"destination": row.destination,
|
|
"quantity": row.quantity,
|
|
"calendarKind": row.calendar_kind or "acid",
|
|
"documentType": row.document_type,
|
|
"sourceRecordId": row.source_record_id,
|
|
"remark": row.remark,
|
|
"createTime": row.create_time,
|
|
"updateTime": row.update_time,
|
|
}
|
|
|
|
|
|
def _query(db: Session, tenant_no: str, user: DbUserContext, model=PlDispatchCalendar):
|
|
query = db.query(model).filter(model.tenant_no == tenant_no, model.del_flag == 0)
|
|
return apply_pl_scope(query, model, user, db, tenant_no)
|
|
|
|
|
|
def _month_bounds(month: str):
|
|
if not re.fullmatch(r"[0-9]{4}-[0-9]{2}", month or ""):
|
|
raise HTTPException(status_code=422, detail="月份须为 YYYY-MM")
|
|
year, number = map(int, month.split("-"))
|
|
if not 1900 <= year <= 2100 or not 1 <= number <= 12:
|
|
raise HTTPException(status_code=422, detail="月份须在 1900-01 至 2100-12 范围内")
|
|
return date(year, number, 1), date(year + (number == 12), number % 12 + 1, 1)
|
|
|
|
|
|
def list_records(db: Session, tenant_no: str, user: DbUserContext, *, month: str,
|
|
vehicle_id: Optional[str] = None, calendar_kind: Optional[str] = None,
|
|
page: int = 1, page_size: int = 500) -> dict:
|
|
start, end = _month_bounds(month)
|
|
query = _query(db, tenant_no, user).filter(
|
|
PlDispatchCalendar.dispatch_date >= start, PlDispatchCalendar.dispatch_date < end,
|
|
PlDispatchCalendar.calendar_kind == _calendar_kind(calendar_kind),
|
|
)
|
|
if vehicle_id:
|
|
query = query.filter(PlDispatchCalendar.vehicle_id == vehicle_id)
|
|
total = query.count()
|
|
rows = query.order_by(PlDispatchCalendar.create_time.desc(), PlDispatchCalendar.id.desc()).offset(
|
|
(page - 1) * page_size
|
|
).limit(page_size).all()
|
|
return {"list": [_row_to_dict(row) for row in rows], "total": total}
|
|
|
|
|
|
def get_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str,
|
|
*, for_update: bool = False) -> PlDispatchCalendar:
|
|
query = _query(db, tenant_no, user).filter(PlDispatchCalendar.id == record_id)
|
|
if for_update:
|
|
# 来源行锁串行化同一来源的更新/删除；刷新会话内缓存，避免并发覆盖关联状态。
|
|
query = query.populate_existing().with_for_update()
|
|
row = query.first()
|
|
if row is None:
|
|
raise HTTPException(status_code=404, detail="派车记录不存在或无权访问")
|
|
return row
|
|
|
|
|
|
def _resolve_snapshots(db: Session, tenant_no: str, user: DbUserContext, body: CreateBody) -> dict:
|
|
vehicle = _query(db, tenant_no, user, PlVehicle).filter(PlVehicle.id == body.vehicle_id).first()
|
|
if vehicle is None:
|
|
raise HTTPException(status_code=422, detail="车辆不存在、已删除或无权访问，请重新选择")
|
|
product = (
|
|
_query(db, tenant_no, user, CrmProduct)
|
|
.filter(CrmProduct.id == body.product_id, CrmProduct.status == "on")
|
|
.first()
|
|
)
|
|
if product is None:
|
|
raise HTTPException(status_code=422, detail="产品不存在、已下架、已删除或无权访问，请重新选择")
|
|
return {"license_plate": vehicle.license_plate, "product_name": product.product_name}
|
|
|
|
|
|
def _site_document_type(origin: str, destination: str) -> Optional[str]:
|
|
outbound = any(word in (origin or "") for word in ("山峡", "云栋"))
|
|
inbound = any(word in (destination or "") for word in ("山峡", "云栋"))
|
|
if outbound and inbound:
|
|
return "出入库计划"
|
|
if outbound:
|
|
return "出库计划"
|
|
if inbound:
|
|
return "入库计划"
|
|
return None
|
|
|
|
|
|
def _sync_site_record(db: Session, tenant_no: str, user: DbUserContext,
|
|
source: PlDispatchCalendar) -> None:
|
|
"""在来源事务和行锁内维护唯一现场记录；不为历史来源补录。"""
|
|
if not source.site_sync_enabled or source.calendar_kind not in ("acid", "alkali", "other"):
|
|
return
|
|
# 包含已软删除关联行，以便重新命中时恢复原 ID 和创建信息。
|
|
query = db.query(PlDispatchCalendar).filter(
|
|
PlDispatchCalendar.tenant_no == tenant_no,
|
|
PlDispatchCalendar.source_record_id == source.id,
|
|
PlDispatchCalendar.calendar_kind == "site",
|
|
)
|
|
target = apply_pl_scope(query, PlDispatchCalendar, user, db, tenant_no).populate_existing().with_for_update().first()
|
|
document_type = None if source.del_flag == 1 else _site_document_type(source.origin, source.destination)
|
|
if document_type is None:
|
|
if target is not None and target.del_flag != 1:
|
|
target.del_flag = 1
|
|
target.update_by = user.id
|
|
return
|
|
if target is None:
|
|
target = PlDispatchCalendar(
|
|
id=generate_uuid(), tenant_no=tenant_no, source_record_id=source.id,
|
|
calendar_kind="site", site_sync_enabled=0, create_by=source.create_by,
|
|
create_dept_id=source.create_dept_id,
|
|
)
|
|
db.add(target)
|
|
for field in ("dispatch_date", "vehicle_id", "license_plate", "product_id", "product_name",
|
|
"origin", "destination", "quantity", "remark"):
|
|
setattr(target, field, getattr(source, field))
|
|
target.document_type = document_type
|
|
target.del_flag = 0
|
|
target.update_by = user.id
|
|
|
|
|
|
def _ensure_manual_record(row: PlDispatchCalendar) -> None:
|
|
if row.source_record_id:
|
|
raise HTTPException(status_code=409, detail="此现场记录由派车日历自动同步，请在来源派车日历修改或删除")
|
|
|
|
|
|
def _retry_deadlock(action):
|
|
"""MySQL 间隙锁竞争可能选择当前事务为死锁牺牲者，回滚后完整重试。"""
|
|
@wraps(action)
|
|
def run(db: Session, *args, **kwargs):
|
|
for attempt in range(3):
|
|
try:
|
|
return action(db, *args, **kwargs)
|
|
except OperationalError as exc:
|
|
db.rollback()
|
|
code = exc.orig.args[0] if getattr(exc.orig, "args", ()) else None
|
|
if code != 1213 or attempt == 2:
|
|
raise
|
|
return run
|
|
|
|
|
|
@_retry_deadlock
|
|
def create_record(db: Session, tenant_no: str, user: DbUserContext, body: CreateBody) -> PlDispatchCalendar:
|
|
snapshots = _resolve_snapshots(db, tenant_no, user, body)
|
|
row = PlDispatchCalendar(
|
|
id=generate_uuid(), tenant_no=tenant_no, create_by=user.id,
|
|
create_dept_id=user.dept_id or "", update_by=user.id,
|
|
site_sync_enabled=1 if body.calendar_kind in ("acid", "alkali", "other") else 0,
|
|
**body.model_dump(), **snapshots,
|
|
)
|
|
try:
|
|
db.add(row)
|
|
db.flush()
|
|
_sync_site_record(db, tenant_no, user, row)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
# ORM 默认值已在 flush/commit 时赋值；提交后不再查询，避免重试已提交的写入。
|
|
return row
|
|
|
|
|
|
@_retry_deadlock
|
|
def update_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str,
|
|
body: UpdateBody) -> PlDispatchCalendar:
|
|
try:
|
|
row = get_record(db, tenant_no, user, record_id, for_update=True)
|
|
_ensure_manual_record(row)
|
|
if body.calendar_kind != (row.calendar_kind or "acid"):
|
|
raise HTTPException(status_code=422, detail="不能把派车计划改到另一本日历")
|
|
values = {**body.model_dump(), **_resolve_snapshots(db, tenant_no, user, body)}
|
|
# 兼容未提交新字段的旧客户端；显式空值仍可清空。
|
|
if "document_type" not in body.model_fields_set:
|
|
values.pop("document_type", None)
|
|
for key, value in values.items():
|
|
setattr(row, key, value)
|
|
row.update_by = user.id
|
|
_sync_site_record(db, tenant_no, user, row)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return row
|
|
|
|
|
|
@_retry_deadlock
|
|
def delete_record(db: Session, tenant_no: str, user: DbUserContext, record_id: str) -> int:
|
|
try:
|
|
row = get_record(db, tenant_no, user, record_id, for_update=True)
|
|
_ensure_manual_record(row)
|
|
row.del_flag = 1
|
|
row.update_by = user.id
|
|
_sync_site_record(db, tenant_no, user, row)
|
|
db.commit()
|
|
except Exception:
|
|
db.rollback()
|
|
raise
|
|
return 1
|
|
|
|
|
|
def _list_options(db: Session, tenant_no: str, user: DbUserContext, model, name_field,
|
|
alias: str, q: Optional[str], page: int, page_size: int) -> dict:
|
|
query = _query(db, tenant_no, user, model)
|
|
keyword = (q or "").strip()
|
|
if keyword:
|
|
query = query.filter(name_field.contains(keyword, autoescape=True))
|
|
total = query.count()
|
|
rows = query.order_by(model.create_time.desc(), model.id.desc()).offset(
|
|
(page - 1) * page_size
|
|
).limit(page_size).all()
|
|
return {"list": [{"id": row.id, alias: getattr(row, name_field.key)} for row in rows], "total": total}
|
|
|
|
|
|
def list_vehicles(db: Session, tenant_no: str, user: DbUserContext, *, q: Optional[str] = None,
|
|
page: int = 1, page_size: int = 500) -> dict:
|
|
return _list_options(db, tenant_no, user, PlVehicle, PlVehicle.license_plate,
|
|
"licensePlate", q, page, page_size)
|
|
|
|
|
|
def list_products(db: Session, tenant_no: str, user: DbUserContext, *, q: Optional[str] = None,
|
|
page: int = 1, page_size: int = 500) -> dict:
|
|
query = _query(db, tenant_no, user, CrmProduct).filter(CrmProduct.status == "on")
|
|
keyword = (q or "").strip()
|
|
if keyword:
|
|
query = query.filter(CrmProduct.product_name.contains(keyword, autoescape=True))
|
|
total = query.count()
|
|
rows = query.order_by(CrmProduct.create_time.desc(), CrmProduct.id.desc()).offset(
|
|
(page - 1) * page_size
|
|
).limit(page_size).all()
|
|
return {"list": [{"id": row.id, "productName": row.product_name} for row in rows], "total": total}
|