Seyyed_arc/hesabixAPI/app/services/distribution_service.py

"""منطق افزونه پخش مویرگی و ویزیتوری."""

from __future__ import annotations

from datetime import date, datetime
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import func, or_
from sqlalchemy.orm import Session, joinedload

from adapters.db.models.distribution import (
	DistributionBusinessSettings,
	DistributionFieldVisit,
	DistributionReturnRequest,
	DistributionRoute,
	DistributionRouteAssignment,
	DistributionRouteStop,
	DistributionTerritory,
)
from adapters.db.models.document import Document
from adapters.db.models.person import Person
from adapters.db.models.product import Product
from app.core.auth_dependency import AuthContext
from app.core.business_calendar import business_today
from app.core.distribution_plugin_dependency import check_distribution_plugin_active
from app.core.responses import ApiError
from app.services.distribution_geo import person_coords


def _ensure_plugin(db: Session, business_id: int) -> None:
	if not check_distribution_plugin_active(db, business_id):
		raise ApiError(
			"DISTRIBUTION_PLUGIN_NOT_ACTIVE",
			"Distribution field sales add-on is not active for this business.",
			http_status=403,
			details={"plugin_code": "distribution", "required_action": "activate_plugin", "marketplace_url": "/marketplace"},
		)


def _scope_visit_user_id(ctx: AuthContext, business_id: int) -> Optional[int]:
	"""فیلتر بازدیدها: None = همهٔ تیم، وگرنه فقط ویزیت‌های خود کاربر."""
	if ctx.is_superadmin():
		return None
	if ctx.db and ctx.is_business_owner(business_id):
		return None
	if ctx.has_business_permission("distribution", "manage"):
		return None
	if ctx.has_business_permission("distribution", "reports_team"):
		return None
	return ctx.get_user_id()


def _can_see_full_distribution_catalog(ctx: AuthContext, business_id: int) -> bool:
	if ctx.is_superadmin():
		return True
	if ctx.db and ctx.is_business_owner(business_id):
		return True
	if ctx.has_business_permission("distribution", "manage"):
		return True
	if ctx.has_business_permission("distribution", "reports_team"):
		return True
	return False


def get_distribution_settings_row(db: Session, business_id: int) -> Optional[DistributionBusinessSettings]:
	return (
		db.query(DistributionBusinessSettings)
		.filter(DistributionBusinessSettings.business_id == business_id)
		.first()
	)


def get_or_create_distribution_settings(db: Session, business_id: int) -> DistributionBusinessSettings:
	row = get_distribution_settings_row(db, business_id)
	if row:
		return row
	row = DistributionBusinessSettings(
		business_id=business_id,
		shared_routing_catalog=False,
		require_visit_in_daily_plan=False,
		created_at=datetime.utcnow(),
		updated_at=datetime.utcnow(),
	)
	db.add(row)
	db.commit()
	db.refresh(row)
	return row


def settings_for_read(db: Session, business_id: int) -> Dict[str, Any]:
	"""خواندن تنظیمات بدون ایجاد رکورد در دیتابیس."""
	row = get_distribution_settings_row(db, business_id)
	if not row:
		from app.services.distribution_phase3_service import extend_settings_dict

		return extend_settings_dict(None)
	return settings_to_dict(row)


def settings_to_dict(s: DistributionBusinessSettings) -> Dict[str, Any]:
	from app.services.distribution_phase3_service import extend_settings_dict

	return extend_settings_dict(s)


def _use_strict_catalog_for_field_user(db: Session, business_id: int, ctx: AuthContext) -> bool:
	if _can_see_full_distribution_catalog(ctx, business_id):
		return False
	s = get_or_create_distribution_settings(db, business_id)
	return not bool(s.shared_routing_catalog)


def _assigned_route_ids_for_user(
	db: Session,
	business_id: int,
	user_id: int,
	on_date: Optional[date] = None,
) -> List[int]:
	q = db.query(DistributionRouteAssignment.route_id).filter(
		DistributionRouteAssignment.business_id == business_id,
		DistributionRouteAssignment.user_id == user_id,
	)
	if on_date is not None:
		q = q.filter(
			DistributionRouteAssignment.valid_from <= on_date,
			or_(DistributionRouteAssignment.valid_to.is_(None), DistributionRouteAssignment.valid_to >= on_date),
		)
	rows = q.distinct().all()
	return [int(r[0]) for r in rows]


def _validate_document_for_business(db: Session, business_id: int, document_id: int, person_id: Optional[int] = None) -> None:
	doc = db.query(Document).filter(Document.id == document_id, Document.business_id == business_id).first()
	if not doc:
		raise ApiError("VALIDATION_ERROR", "document_id not found in this business", http_status=400)
	if person_id is not None:
		ex = doc.extra_info or {}
		doc_person = ex.get("person_id")
		if doc_person is not None and int(doc_person) != int(person_id):
			raise ApiError("VALIDATION_ERROR", "document does not belong to this person", http_status=400)


def _validate_return_lines(db: Session, business_id: int, lines: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
	out: List[Dict[str, Any]] = []
	for i, ln in enumerate(lines):
		if not isinstance(ln, dict):
			raise ApiError("VALIDATION_ERROR", f"lines[{i}] must be object", http_status=400)
		pid = int(ln.get("product_id") or 0)
		if pid <= 0:
			raise ApiError("VALIDATION_ERROR", f"lines[{i}].product_id required", http_status=400)
		product = db.query(Product).filter(Product.id == pid, Product.business_id == business_id).first()
		if not product:
			raise ApiError("NOT_FOUND", f"Product {pid} not found", http_status=404)
		try:
			qty = float(ln.get("quantity") or 0)
		except (TypeError, ValueError):
			raise ApiError("VALIDATION_ERROR", f"lines[{i}].quantity invalid", http_status=400) from None
		if qty <= 0:
			raise ApiError("VALIDATION_ERROR", f"lines[{i}].quantity must be positive", http_status=400)
		out.append(
			{
				"product_id": pid,
				"product_name": getattr(product, "name", None) or getattr(product, "title", None),
				"quantity": qty,
				"reason": (str(ln.get("reason")).strip() if ln.get("reason") else None),
				"reason_code": (str(ln.get("reason_code")).strip().lower() if ln.get("reason_code") else None),
				"unit": (str(ln.get("unit")).strip() if ln.get("unit") else None),
			}
		)
	return out


def _cap_return_lines_against_source(
	db: Session,
	business_id: int,
	source_document_id: int,
	lines: List[Dict[str, Any]],
) -> None:
	"""اگر خط مرجوعی از مقدار فروخته‌شده در فاکتور منبع بیشتر باشد، خطا بده."""
	from adapters.db.models.document_line import DocumentLine

	sold: Dict[int, float] = {}
	for dl in (
		db.query(DocumentLine)
		.filter(DocumentLine.document_id == source_document_id, DocumentLine.product_id.isnot(None))
		.all()
	):
		pid = int(dl.product_id)
		sold[pid] = sold.get(pid, 0.0) + float(dl.quantity or 0)
	if not sold:
		return
	for i, ln in enumerate(lines):
		pid = int(ln["product_id"])
		qty = float(ln["quantity"])
		max_qty = sold.get(pid)
		if max_qty is None:
			raise ApiError(
				"VALIDATION_ERROR",
				f"lines[{i}]: product {pid} not on source invoice",
				http_status=400,
			)
		if qty > max_qty + 1e-9:
			raise ApiError(
				"VALIDATION_ERROR",
				f"lines[{i}]: quantity {qty} exceeds sold {max_qty} on source invoice",
				http_status=400,
			)


def _ensure_route_allowed_for_strict(db: Session, business_id: int, ctx: AuthContext, route_id: int) -> None:
	if not _use_strict_catalog_for_field_user(db, business_id, ctx):
		return
	uid = ctx.get_user_id()
	if uid is None:
		return
	if route_id not in _assigned_route_ids_for_user(db, business_id, uid, on_date=business_today(business_id)):
		raise ApiError("FORBIDDEN", "مسیر به شما تخصیص داده نشده است.", http_status=403)



def territory_to_dict(t: DistributionTerritory) -> Dict[str, Any]:
	return {
		"id": t.id,
		"code": t.code,
		"name": t.name,
		"description": t.description,
		"is_active": t.is_active,
		"created_at": t.created_at.isoformat() if t.created_at else None,
		"updated_at": t.updated_at.isoformat() if t.updated_at else None,
	}


def route_to_dict(r: DistributionRoute, territory_name: Optional[str] = None) -> Dict[str, Any]:
	return {
		"id": r.id,
		"code": r.code,
		"name": r.name,
		"description": r.description,
		"is_active": r.is_active,
		"territory_id": r.territory_id,
		"territory_name": territory_name,
		"created_at": r.created_at.isoformat() if r.created_at else None,
		"updated_at": r.updated_at.isoformat() if r.updated_at else None,
	}


def visit_to_dict(v: DistributionFieldVisit, person_name: Optional[str] = None) -> Dict[str, Any]:
	def _num(val: Any) -> Any:
		if val is None:
			return None
		try:
			return float(val)
		except Exception:
			return val

	return {
		"id": v.id,
		"person_id": v.person_id,
		"person_name": person_name,
		"user_id": v.user_id,
		"route_id": v.route_id,
		"route_stop_id": v.route_stop_id,
		"status": v.status,
		"started_at": v.started_at.isoformat() if v.started_at else None,
		"ended_at": v.ended_at.isoformat() if v.ended_at else None,
		"outcome": v.outcome,
		"no_order_reason": v.no_order_reason,
		"document_id": v.document_id,
		"deal_id": v.deal_id,
		"crm_activity_id": v.crm_activity_id,
		"notes": v.notes,
		"extra_info": v.extra_info,
		"start_latitude": _num(getattr(v, "start_latitude", None)),
		"start_longitude": _num(getattr(v, "start_longitude", None)),
		"end_latitude": _num(getattr(v, "end_latitude", None)),
		"end_longitude": _num(getattr(v, "end_longitude", None)),
		"checklist_answers": getattr(v, "checklist_answers", None),
		"shelf_photo_file_id": getattr(v, "shelf_photo_file_id", None),
		"supervisor_user_id": getattr(v, "supervisor_user_id", None),
		"no_order_reason_code": getattr(v, "no_order_reason_code", None),
		"is_carried_over": bool(getattr(v, "is_carried_over", False)),
		"pod_signature_file_id": getattr(v, "pod_signature_file_id", None),
		"time_in_store_seconds": (
			int((v.ended_at - v.started_at).total_seconds())
			if v.started_at and v.ended_at
			else None
		),
	}


def get_summary(db: Session, business_id: int, ctx: AuthContext) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	scope_uid = _scope_visit_user_id(ctx, business_id)

	vq = db.query(DistributionFieldVisit).filter(DistributionFieldVisit.business_id == business_id)
	if scope_uid is not None:
		vq = vq.filter(DistributionFieldVisit.user_id == scope_uid)

	today = business_today(business_id)
	start_today = datetime.combine(today, datetime.min.time())
	end_today = datetime.combine(today, datetime.max.time())
	visits_today = vq.filter(
		DistributionFieldVisit.started_at >= start_today,
		DistributionFieldVisit.started_at <= end_today,
	).count()

	completed_today = vq.filter(
		DistributionFieldVisit.started_at >= start_today,
		DistributionFieldVisit.started_at <= end_today,
		DistributionFieldVisit.status == "completed",
	).count()

	pending_returns = db.query(func.count(DistributionReturnRequest.id)).filter(
		DistributionReturnRequest.business_id == business_id,
		DistributionReturnRequest.status == "pending",
	).scalar()
	if scope_uid is not None:
		pending_returns = (
			db.query(func.count(DistributionReturnRequest.id))
			.filter(
				DistributionReturnRequest.business_id == business_id,
				DistributionReturnRequest.status == "pending",
				DistributionReturnRequest.created_by_user_id == scope_uid,
			)
			.scalar()
		)

	strict_cat = _use_strict_catalog_for_field_user(db, business_id, ctx)
	uid_summary = ctx.get_user_id()
	if strict_cat and uid_summary is not None:
		assigned = _assigned_route_ids_for_user(db, business_id, uid_summary, on_date=business_today(business_id))
		if assigned:
			routes_active = (
				db.query(func.count(DistributionRoute.id))
				.filter(
					DistributionRoute.business_id == business_id,
					DistributionRoute.is_active == True,  # noqa: E712
					DistributionRoute.id.in_(assigned),
				)
				.scalar()
			)
		else:
			routes_active = 0
	else:
		routes_active = (
			db.query(func.count(DistributionRoute.id))
			.filter(DistributionRoute.business_id == business_id, DistributionRoute.is_active == True)  # noqa: E712
			.scalar()
		)

	return {
		"visits_today": visits_today,
		"completed_visits_today": completed_today,
		"pending_return_requests": int(pending_returns or 0),
		"active_routes": int(routes_active or 0),
		"distribution_settings": settings_for_read(db, business_id),
	}


def list_territories(db: Session, business_id: int, ctx: AuthContext) -> List[Dict[str, Any]]:
	_ensure_plugin(db, business_id)
	q = db.query(DistributionTerritory).filter(DistributionTerritory.business_id == business_id)
	if _use_strict_catalog_for_field_user(db, business_id, ctx):
		uid = ctx.get_user_id()
		if uid is None:
			return []
		route_ids = _assigned_route_ids_for_user(db, business_id, uid, on_date=business_today(business_id))
		if not route_ids:
			return []
		tids = (
			db.query(DistributionRoute.territory_id)
			.filter(DistributionRoute.business_id == business_id, DistributionRoute.id.in_(route_ids))
			.distinct()
			.all()
		)
		tid_list = [int(x[0]) for x in tids if x[0] is not None]
		if not tid_list:
			return []
		q = q.filter(DistributionTerritory.id.in_(tid_list))
	return [territory_to_dict(t) for t in q.order_by(DistributionTerritory.code.asc()).all()]


def create_territory(db: Session, business_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	code = str(payload.get("code") or "").strip()
	name = str(payload.get("name") or "").strip()
	if not code or not name:
		raise ApiError("VALIDATION_ERROR", "code and name are required", http_status=400)
	exists = (
		db.query(DistributionTerritory)
		.filter(DistributionTerritory.business_id == business_id, DistributionTerritory.code == code)
		.first()
	)
	if exists:
		raise ApiError("DUPLICATE", "Territory code already exists", http_status=400)
	row = DistributionTerritory(
		business_id=business_id,
		code=code[:50],
		name=name[:255],
		description=(payload.get("description") or None),
		is_active=bool(payload.get("is_active", True)),
	)
	db.add(row)
	db.commit()
	db.refresh(row)
	return territory_to_dict(row)


def update_territory(db: Session, business_id: int, territory_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	row = (
		db.query(DistributionTerritory)
		.filter(DistributionTerritory.id == territory_id, DistributionTerritory.business_id == business_id)
		.first()
	)
	if not row:
		raise ApiError("NOT_FOUND", "Territory not found", http_status=404)
	if "name" in payload:
		row.name = str(payload["name"])[:255]
	if "description" in payload:
		row.description = payload.get("description")
	if "is_active" in payload:
		row.is_active = bool(payload["is_active"])
	row.updated_at = datetime.utcnow()
	db.commit()
	db.refresh(row)
	return territory_to_dict(row)


def delete_territory(db: Session, business_id: int, territory_id: int) -> None:
	_ensure_plugin(db, business_id)
	row = (
		db.query(DistributionTerritory)
		.filter(DistributionTerritory.id == territory_id, DistributionTerritory.business_id == business_id)
		.first()
	)
	if not row:
		raise ApiError("NOT_FOUND", "Territory not found", http_status=404)
	db.query(DistributionRoute).filter(DistributionRoute.territory_id == territory_id).update({"territory_id": None})
	db.delete(row)
	db.commit()


def list_routes(db: Session, business_id: int, ctx: AuthContext) -> List[Dict[str, Any]]:
	_ensure_plugin(db, business_id)
	q = (
		db.query(DistributionRoute)
		.options(joinedload(DistributionRoute.territory))
		.filter(DistributionRoute.business_id == business_id)
	)
	if _use_strict_catalog_for_field_user(db, business_id, ctx):
		uid = ctx.get_user_id()
		if uid is None:
			return []
		route_ids = _assigned_route_ids_for_user(db, business_id, uid, on_date=business_today(business_id))
		if not route_ids:
			return []
		q = q.filter(DistributionRoute.id.in_(route_ids))
	rows = q.order_by(DistributionRoute.code.asc()).all()
	out = []
	for r in rows:
		tname = r.territory.name if r.territory else None
		out.append(route_to_dict(r, tname))
	return out


def create_route(db: Session, business_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	code = str(payload.get("code") or "").strip()
	name = str(payload.get("name") or "").strip()
	if not code or not name:
		raise ApiError("VALIDATION_ERROR", "code and name are required", http_status=400)
	if (
		db.query(DistributionRoute)
		.filter(DistributionRoute.business_id == business_id, DistributionRoute.code == code)
		.first()
	):
		raise ApiError("DUPLICATE", "Route code already exists", http_status=400)
	tid = payload.get("territory_id")
	if tid is not None:
		tid = int(tid)
		check = db.query(DistributionTerritory).filter(DistributionTerritory.id == tid, DistributionTerritory.business_id == business_id).first()
		if not check:
			raise ApiError("VALIDATION_ERROR", "Invalid territory_id", http_status=400)
	row = DistributionRoute(
		business_id=business_id,
		territory_id=tid,
		code=code[:50],
		name=name[:255],
		description=(payload.get("description") or None),
		is_active=bool(payload.get("is_active", True)),
	)
	db.add(row)
	db.commit()
	db.refresh(row)
	tname = None
	if row.territory_id:
		tr = db.query(DistributionTerritory).filter(DistributionTerritory.id == row.territory_id).first()
		tname = tr.name if tr else None
	return route_to_dict(row, tname)


def delete_route(db: Session, business_id: int, route_id: int) -> None:
	_ensure_plugin(db, business_id)
	row = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not row:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	in_progress = (
		db.query(DistributionFieldVisit.id)
		.filter(
			DistributionFieldVisit.business_id == business_id,
			DistributionFieldVisit.route_id == route_id,
			DistributionFieldVisit.status == "in_progress",
		)
		.first()
	)
	if in_progress:
		raise ApiError("CONFLICT", "Cannot delete route with visits in progress", http_status=409)
	db.delete(row)
	db.commit()


def update_route(db: Session, business_id: int, route_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	row = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not row:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	if "name" in payload:
		row.name = str(payload["name"])[:255]
	if "description" in payload:
		row.description = payload.get("description")
	if "is_active" in payload:
		row.is_active = bool(payload["is_active"])
	if "territory_id" in payload:
		tid = payload.get("territory_id")
		if tid is None:
			row.territory_id = None
		else:
			check = db.query(DistributionTerritory).filter(DistributionTerritory.id == int(tid), DistributionTerritory.business_id == business_id).first()
			if not check:
				raise ApiError("VALIDATION_ERROR", "Invalid territory_id", http_status=400)
			row.territory_id = int(tid)
	row.updated_at = datetime.utcnow()
	db.commit()
	db.refresh(row)
	tname = None
	if row.territory_id:
		t = db.query(DistributionTerritory).filter(DistributionTerritory.id == row.territory_id).first()
		tname = t.name if t else None
	return route_to_dict(row, tname)


def list_route_stops(db: Session, business_id: int, route_id: int, ctx: AuthContext) -> List[Dict[str, Any]]:
	_ensure_plugin(db, business_id)
	rt = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not rt:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	_ensure_route_allowed_for_strict(db, business_id, ctx, route_id)
	stops = (
		db.query(DistributionRouteStop)
		.filter(DistributionRouteStop.route_id == route_id)
		.order_by(DistributionRouteStop.sort_order.asc(), DistributionRouteStop.id.asc())
		.all()
	)
	person_ids = [s.person_id for s in stops]
	persons = {}
	if person_ids:
		for p in db.query(Person).filter(Person.id.in_(person_ids)).all():
			persons[p.id] = (p.alias_name or "").strip() or str(p.id)
	out = []
	for s in stops:
		out.append(
			{
				"id": s.id,
				"person_id": s.person_id,
				"person_name": persons.get(s.person_id),
				"sort_order": s.sort_order,
				"weekday": s.weekday,
				"frequency": getattr(s, "frequency", None) or "weekly",
				"cycle_offset": int(getattr(s, "cycle_offset", 0) or 0),
				"customer_class": getattr(s, "customer_class", None),
				"notes": s.notes,
			}
		)
	return out


def upsert_route_stop(db: Session, business_id: int, route_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	rt = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not rt:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	person_id = int(payload.get("person_id") or 0)
	if person_id <= 0:
		raise ApiError("VALIDATION_ERROR", "person_id required", http_status=400)
	person = db.query(Person).filter(Person.id == person_id, Person.business_id == business_id).first()
	if not person:
		raise ApiError("NOT_FOUND", "Person not found in this business", http_status=404)
	stop_id = payload.get("id")
	weekday = payload.get("weekday")
	if weekday is not None and weekday != "":
		weekday = int(weekday)
		if weekday < 0 or weekday > 6:
			raise ApiError("VALIDATION_ERROR", "weekday must be 0-6 or null", http_status=400)
	else:
		weekday = None
	sort_order = int(payload.get("sort_order") or 0)
	frequency = str(payload.get("frequency") or "weekly").strip().lower()
	if frequency not in ("weekly", "biweekly", "monthly"):
		frequency = "weekly"
	cycle_offset = max(0, int(payload.get("cycle_offset") or 0))
	customer_class = str(payload.get("customer_class") or "").strip().upper() or None
	if customer_class and customer_class not in ("A", "B", "C"):
		raise ApiError("VALIDATION_ERROR", "customer_class must be A|B|C", http_status=400)
	if stop_id:
		row = db.query(DistributionRouteStop).filter(DistributionRouteStop.id == int(stop_id), DistributionRouteStop.route_id == route_id).first()
		if not row:
			raise ApiError("NOT_FOUND", "Stop not found", http_status=404)
		row.person_id = person_id
		row.sort_order = sort_order
		row.weekday = weekday
		row.notes = payload.get("notes")
		row.frequency = frequency
		row.cycle_offset = cycle_offset
		row.customer_class = customer_class
		row.updated_at = datetime.utcnow()
	else:
		row = DistributionRouteStop(
			route_id=route_id,
			person_id=person_id,
			sort_order=sort_order,
			weekday=weekday,
			notes=payload.get("notes"),
			frequency=frequency,
			cycle_offset=cycle_offset,
			customer_class=customer_class,
		)
		db.add(row)
	db.commit()
	db.refresh(row)
	pname = (person.alias_name or "").strip() or str(person.id)
	return {
		"id": row.id,
		"person_id": row.person_id,
		"person_name": pname,
		"sort_order": row.sort_order,
		"weekday": row.weekday,
		"frequency": getattr(row, "frequency", None) or "weekly",
		"cycle_offset": int(getattr(row, "cycle_offset", 0) or 0),
		"customer_class": getattr(row, "customer_class", None),
		"notes": row.notes,
	}


def delete_route_stop(db: Session, business_id: int, route_id: int, stop_id: int) -> None:
	_ensure_plugin(db, business_id)
	rt = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not rt:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	row = db.query(DistributionRouteStop).filter(DistributionRouteStop.id == stop_id, DistributionRouteStop.route_id == route_id).first()
	if not row:
		raise ApiError("NOT_FOUND", "Stop not found", http_status=404)
	db.delete(row)
	db.commit()


def list_assignments(db: Session, business_id: int, route_id: Optional[int], ctx: AuthContext) -> List[Dict[str, Any]]:
	_ensure_plugin(db, business_id)
	q = db.query(DistributionRouteAssignment).filter(DistributionRouteAssignment.business_id == business_id)
	if route_id:
		q = q.filter(DistributionRouteAssignment.route_id == route_id)
	if _use_strict_catalog_for_field_user(db, business_id, ctx):
		uid = ctx.get_user_id()
		if uid is None:
			return []
		q = q.filter(DistributionRouteAssignment.user_id == uid)
	rows = q.order_by(DistributionRouteAssignment.valid_from.desc()).all()
	from app.services.distribution_documents import user_label

	out = []
	for a in rows:
		out.append(
			{
				"id": a.id,
				"route_id": a.route_id,
				"user_id": a.user_id,
				"user_name": user_label(db, a.user_id),
				"valid_from": a.valid_from.isoformat() if a.valid_from else None,
				"valid_to": a.valid_to.isoformat() if a.valid_to else None,
				"created_at": a.created_at.isoformat() if a.created_at else None,
			}
		)
	return out


def create_assignment(db: Session, business_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	route_id = int(payload.get("route_id") or 0)
	user_id = int(payload.get("user_id") or 0)
	if route_id <= 0 or user_id <= 0:
		raise ApiError("VALIDATION_ERROR", "route_id and user_id required", http_status=400)
	rt = db.query(DistributionRoute).filter(DistributionRoute.id == route_id, DistributionRoute.business_id == business_id).first()
	if not rt:
		raise ApiError("NOT_FOUND", "Route not found", http_status=404)
	vf = payload.get("valid_from")
	if not vf:
		raise ApiError("VALIDATION_ERROR", "valid_from required (YYYY-MM-DD)", http_status=400)
	valid_from = date.fromisoformat(str(vf)[:10])
	valid_to = None
	if payload.get("valid_to"):
		valid_to = date.fromisoformat(str(payload["valid_to"])[:10])
	row = DistributionRouteAssignment(
		business_id=business_id,
		route_id=route_id,
		user_id=user_id,
		valid_from=valid_from,
		valid_to=valid_to,
	)
	db.add(row)
	db.commit()
	db.refresh(row)
	return {
		"id": row.id,
		"route_id": row.route_id,
		"user_id": row.user_id,
		"valid_from": row.valid_from.isoformat(),
		"valid_to": row.valid_to.isoformat() if row.valid_to else None,
	}


def delete_assignment(db: Session, business_id: int, assignment_id: int) -> None:
	_ensure_plugin(db, business_id)
	row = (
		db.query(DistributionRouteAssignment)
		.filter(DistributionRouteAssignment.id == assignment_id, DistributionRouteAssignment.business_id == business_id)
		.first()
	)
	if not row:
		raise ApiError("NOT_FOUND", "Assignment not found", http_status=404)
	db.delete(row)
	db.commit()


def _weekday_matches(plan_date: date, weekday: Optional[int]) -> bool:
	if weekday is None:
		return True
	return int(weekday) == int(plan_date.weekday())


def get_daily_plan(db: Session, business_id: int, target_user_id: int, plan_date: date) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	assignments = (
		db.query(DistributionRouteAssignment)
		.options(joinedload(DistributionRouteAssignment.route))
		.filter(
			DistributionRouteAssignment.business_id == business_id,
			DistributionRouteAssignment.user_id == target_user_id,
			DistributionRouteAssignment.valid_from <= plan_date,
			or_(DistributionRouteAssignment.valid_to.is_(None), DistributionRouteAssignment.valid_to >= plan_date),
		)
		.all()
	)
	items: List[Dict[str, Any]] = []
	for asn in assignments:
		route = asn.route
		if not route or not route.is_active:
			continue
		stops = (
			db.query(DistributionRouteStop)
			.filter(DistributionRouteStop.route_id == route.id)
			.order_by(DistributionRouteStop.sort_order.asc())
			.all()
		)
		for s in stops:
			if not _weekday_matches(plan_date, s.weekday):
				continue
			person = db.query(Person).filter(Person.id == s.person_id).first()
			pname = (person.alias_name or "").strip() if person else None
			plats = person_coords(person) if person else (None, None)
			# ترتیب روزمحور از override مسیر (در صورت وجود)
			day_sort = s.sort_order
			overrides = getattr(route, "plan_sort_overrides", None) or {}
			day_map = overrides.get(plan_date.isoformat()) if isinstance(overrides, dict) else None
			if isinstance(day_map, dict) and str(s.id) in day_map:
				try:
					day_sort = int(day_map[str(s.id)])
				except (TypeError, ValueError):
					day_sort = s.sort_order
			mobile = None
			if person is not None:
				for cand in (getattr(person, "mobile", None), getattr(person, "phone", None), getattr(person, "mobile_2", None)):
					if cand and str(cand).strip():
						mobile = str(cand).strip()
						break
			items.append(
				{
					"route_id": route.id,
					"route_code": route.code,
					"route_name": route.name,
					"stop_id": s.id,
					"sort_order": day_sort,
					"catalog_sort_order": s.sort_order,
					"person_id": s.person_id,
					"person_name": pname or str(s.person_id),
					"person_mobile": mobile,
					"weekday": s.weekday,
					"latitude": plats[0],
					"longitude": plats[1],
					"day_optimized": isinstance(day_map, dict) and str(s.id) in day_map,
				}
			)
	items.sort(key=lambda x: (x["route_code"], x["sort_order"], x["person_id"]))
	from app.services.distribution_field_ops_service import enrich_daily_plan

	return enrich_daily_plan(db, business_id, target_user_id, plan_date, items)


def _active_visit(db: Session, business_id: int, user_id: int) -> Optional[DistributionFieldVisit]:
	return (
		db.query(DistributionFieldVisit)
		.filter(
			DistributionFieldVisit.business_id == business_id,
			DistributionFieldVisit.user_id == user_id,
			DistributionFieldVisit.status == "in_progress",
		)
		.order_by(DistributionFieldVisit.started_at.desc())
		.first()
	)


def start_visit(
	db: Session,
	business_id: int,
	user_id: int,
	payload: Dict[str, Any],
	ctx: Optional[AuthContext] = None,
) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	if _active_visit(db, business_id, user_id):
		raise ApiError("CONFLICT", "یک ویزیت باز دارید؛ ابتدا آن را تکمیل کنید.", http_status=409)
	person_id = int(payload.get("person_id") or 0)
	if person_id <= 0:
		raise ApiError("VALIDATION_ERROR", "person_id required", http_status=400)
	person = db.query(Person).filter(Person.id == person_id, Person.business_id == business_id).first()
	if not person:
		raise ApiError("NOT_FOUND", "Person not found", http_status=404)
	today = business_today(business_id)
	settings_sv = get_or_create_distribution_settings(db, business_id)
	if settings_sv.require_visit_in_daily_plan:
		_plan = get_daily_plan(db, business_id, user_id, today)
		_allowed_pids = {int(it["person_id"]) for it in _plan["items"]}
		if person_id not in _allowed_pids:
			raise ApiError(
				"VALIDATION_ERROR",
				"این شخص در برنامهٔ روز ویزیت شما نیست.",
				http_status=400,
			)
	route_id = payload.get("route_id")
	route_stop_id = payload.get("route_stop_id")
	if ctx is not None and route_id:
		_ensure_route_allowed_for_strict(db, business_id, ctx, int(route_id))
	if ctx is not None and _use_strict_catalog_for_field_user(db, business_id, ctx):
		allowed_routes = _assigned_route_ids_for_user(db, business_id, user_id, on_date=today)
		if not allowed_routes:
			raise ApiError("FORBIDDEN", "هیچ مسیری به شما تخصیص داده نشده است.", http_status=403)
		if route_id and int(route_id) not in allowed_routes:
			raise ApiError("FORBIDDEN", "مسیر به شما تخصیص داده نشده است.", http_status=403)
		if not route_id:
			_plan = get_daily_plan(db, business_id, user_id, today)
			plan_person_routes = {int(it["person_id"]): int(it["route_id"]) for it in _plan["items"]}
			if person_id not in plan_person_routes:
				raise ApiError(
					"FORBIDDEN",
					"این مشتری در مسیرهای تخصیص‌یافتهٔ شما برای امروز نیست.",
					http_status=403,
				)
	_lat = payload.get("start_latitude")
	_lng = payload.get("start_longitude")
	_slat = float(_lat) if _lat is not None and str(_lat).strip() != "" else None
	_slng = float(_lng) if _lng is not None and str(_lng).strip() != "" else None
	from app.services.distribution_phase3_service import validate_geofence_on_start

	want_override = bool(payload.get("geofence_override"))
	if want_override:
		can_override = False
		if ctx is not None:
			can_override = _can_see_full_distribution_catalog(ctx, business_id) or ctx.has_business_permission(
				"distribution", "manage"
			)
		if not can_override:
			raise ApiError(
				"FORBIDDEN",
				"geofence_override requires distribution.manage",
				http_status=403,
			)
		override_reason = str(payload.get("geofence_override_reason") or "").strip()
		if len(override_reason) < 3:
			raise ApiError(
				"VALIDATION_ERROR",
				"geofence_override_reason required (min 3 chars)",
				http_status=400,
			)
	geo = validate_geofence_on_start(
		db,
		business_id,
		person_id,
		_slat,
		_slng,
		allow_override=want_override,
	)
	_extra = payload.get("extra_info")
	if _extra is not None and not isinstance(_extra, dict):
		raise ApiError("VALIDATION_ERROR", "extra_info must be an object", http_status=400)
	_extra = dict(_extra) if isinstance(_extra, dict) else {}
	if geo.get("geofence_warning"):
		_extra["geofence_warning"] = geo["geofence_warning"]
	if geo.get("distance_meters") is not None:
		_extra["geofence_distance_meters"] = geo["distance_meters"]
	if want_override:
		_extra["geofence_override"] = True
		_extra["geofence_override_reason"] = str(payload.get("geofence_override_reason") or "").strip()
		_extra["geofence_override_by"] = user_id
	v = DistributionFieldVisit(
		business_id=business_id,
		person_id=person_id,
		user_id=user_id,
		route_id=int(route_id) if route_id else None,
		route_stop_id=int(route_stop_id) if route_stop_id else None,
		status="in_progress",
		started_at=datetime.utcnow(),
		notes=payload.get("notes"),
		start_latitude=_slat,
		start_longitude=_slng,
		extra_info=_extra or None,
		supervisor_user_id=int(payload["supervisor_user_id"]) if payload.get("supervisor_user_id") else None,
		is_carried_over=bool(payload.get("is_carried_over")),
	)
	db.add(v)
	db.commit()
	db.refresh(v)
	pname = (person.alias_name or "").strip()
	return visit_to_dict(v, pname)


def heartbeat_visit(
	db: Session,
	business_id: int,
	user_id: int,
	visit_id: int,
	latitude: float,
	longitude: float,
) -> Dict[str, Any]:
	"""به‌روزرسانی موقعیت زندهٔ ویزیت در حال انجام برای نقشه تیم."""
	_ensure_plugin(db, business_id)
	v = (
		db.query(DistributionFieldVisit)
		.filter(
			DistributionFieldVisit.id == visit_id,
			DistributionFieldVisit.business_id == business_id,
		)
		.first()
	)
	if not v:
		raise ApiError("NOT_FOUND", "Visit not found", http_status=404)
	if int(v.user_id) != int(user_id):
		raise ApiError("FORBIDDEN", "Only the assigned visitor can update location", http_status=403)
	if v.status != "in_progress":
		raise ApiError("VALIDATION_ERROR", "Visit is not in progress", http_status=400)
	_base = dict(v.extra_info or {})
	_base["live_location"] = {
		"latitude": float(latitude),
		"longitude": float(longitude),
		"updated_at": datetime.utcnow().isoformat() + "Z",
	}
	v.extra_info = _base
	# برای سازگاری با خواننده‌های قدیمی، start را هم به‌روز نگه می‌داریم به‌عنوان آخرین نقطه
	v.start_latitude = float(latitude)
	v.start_longitude = float(longitude)
	v.updated_at = datetime.utcnow()
	try:
		from app.services.distribution_commercial_service import upsert_live_location
		from app.services.distribution_phase3_service import extend_settings_dict

		settings = get_or_create_distribution_settings(db, business_id)
		if bool(extend_settings_dict(settings).get("share_live_location", True)):
			upsert_live_location(db, business_id, user_id, latitude, longitude, visit_id=visit_id, commit=False)
	except Exception:
		pass
	db.commit()
	db.refresh(v)
	return {
		"visit_id": v.id,
		"latitude": float(latitude),
		"longitude": float(longitude),
		"updated_at": _base["live_location"]["updated_at"],
	}


def get_person_credit_summary(db: Session, business_id: int, person_id: int) -> Dict[str, Any]:
	"""خلاصه اعتبار مشتری برای UI میدانی."""
	_ensure_plugin(db, business_id)
	from app.services.credit_service import get_person_credit, get_business_credit_settings
	from app.services.person_service import calculate_person_balance

	person = db.query(Person).filter(Person.id == person_id, Person.business_id == business_id).first()
	if not person:
		raise ApiError("NOT_FOUND", "Person not found", http_status=404)
	credit = get_person_credit(db, business_id, person_id)
	biz_settings = get_business_credit_settings(db, business_id)
	check_enabled = credit.get("credit_check_enabled")
	if check_enabled is None:
		check_enabled = bool(biz_settings.get("is_enabled"))
	balance = None
	debt = None
	try:
		bal, _st = calculate_person_balance(db, person_id)
		if bal is not None:
			balance = float(bal)
			debt = float(-bal) if bal < 0 else 0.0
	except Exception:
		pass
	limit = credit.get("effective_credit_limit")
	available = None
	if check_enabled and limit is not None and debt is not None:
		available = float(limit) - float(debt)
	return {
		"person_id": person_id,
		"person_name": (person.alias_name or "").strip() or str(person_id),
		"credit_check_enabled": bool(check_enabled),
		"credit_limit": limit,
		"balance": balance,
		"current_debt": debt,
		"available_credit": available,
		"blocked": bool(check_enabled and available is not None and available < 0),
	}


def complete_visit(
	db: Session,
	business_id: int,
	user_id: int,
	visit_id: int,
	payload: Dict[str, Any],
	allow_manage_override: bool = False,
) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	v = db.query(DistributionFieldVisit).filter(DistributionFieldVisit.id == visit_id, DistributionFieldVisit.business_id == business_id).first()
	if not v:
		raise ApiError("NOT_FOUND", "Visit not found", http_status=404)
	if v.user_id != user_id and not allow_manage_override:
		raise ApiError("FORBIDDEN", "Only the assigned visitor can complete this visit", http_status=403)
	if v.status != "in_progress":
		raise ApiError("VALIDATION_ERROR", "Visit is not in progress", http_status=400)
	outcome = str(payload.get("outcome") or "").strip()
	if outcome not in ("order", "no_order"):
		raise ApiError("VALIDATION_ERROR", "outcome must be order | no_order", http_status=400)
	reason_code = str(payload.get("no_order_reason_code") or "").strip().lower() or None
	if outcome == "no_order":
		from app.services.distribution_field_ops_service import NO_ORDER_REASONS

		allowed = {r["code"] for r in NO_ORDER_REASONS}
		if reason_code and reason_code not in allowed:
			raise ApiError("VALIDATION_ERROR", "invalid no_order_reason_code", http_status=400)
		if not reason_code and not str(payload.get("no_order_reason") or "").strip():
			raise ApiError("VALIDATION_ERROR", "دلیل عدم سفارش را انتخاب کنید", http_status=400)
	v.status = "completed"
	v.ended_at = datetime.utcnow()
	v.outcome = outcome
	v.no_order_reason = (payload.get("no_order_reason") or None)
	if reason_code:
		v.no_order_reason_code = reason_code
	if payload.get("document_id"):
		doc_id = int(payload["document_id"])
		_validate_document_for_business(db, business_id, doc_id, v.person_id)
		v.document_id = doc_id
	if payload.get("deal_id"):
		from adapters.db.models.crm import Deal

		deal_id = int(payload["deal_id"])
		deal = db.query(Deal).filter(Deal.id == deal_id, Deal.business_id == business_id).first()
		if not deal:
			raise ApiError("VALIDATION_ERROR", "deal_id not found in this business", http_status=400)
		v.deal_id = deal_id
	if payload.get("notes"):
		v.notes = str(payload["notes"])
	if "extra_info" in payload:
		_ex = payload.get("extra_info")
		if _ex is not None and not isinstance(_ex, dict):
			raise ApiError("VALIDATION_ERROR", "extra_info must be an object", http_status=400)
		_base = dict(v.extra_info or {})
		_base.update(_ex or {})
		v.extra_info = _base
	# POD — تأیید تحویل/امضا
	if outcome == "order":
		pod_confirmed = payload.get("pod_confirmed")
		pod_name = str(payload.get("pod_signer_name") or "").strip()
		if pod_confirmed is True:
			if len(pod_name) < 2:
				raise ApiError(
					"VALIDATION_ERROR",
					"pod_signer_name required when pod_confirmed",
					http_status=400,
				)
			settings_pod = get_or_create_distribution_settings(db, business_id)
			sig = str(payload.get("pod_signature_png") or "").strip()
			if getattr(settings_pod, "require_pod_signature", False) and len(sig) < 40 and not payload.get("pod_signature_file_id"):
				raise ApiError("VALIDATION_ERROR", "امضای تحویل الزامی است", http_status=400)
			if getattr(settings_pod, "require_pod_photo", False) and not payload.get("pod_photo_file_id") and not payload.get("shelf_photo_file_id"):
				raise ApiError("VALIDATION_ERROR", "عکس تحویل الزامی است", http_status=400)
			_base = dict(v.extra_info or {})
			_base["pod"] = {
				"confirmed": True,
				"signer_name": pod_name[:255],
				"note": (str(payload.get("pod_note") or "").strip()[:500] or None),
				"confirmed_at": datetime.utcnow().isoformat() + "Z",
				"confirmed_by_user_id": user_id,
				"signature_png": sig[:250000] if sig else None,
				"photo_file_id": payload.get("pod_photo_file_id") or payload.get("shelf_photo_file_id"),
			}
			v.extra_info = _base
			if payload.get("pod_signature_file_id"):
				v.pod_signature_file_id = int(payload["pod_signature_file_id"])
	_elat = payload.get("end_latitude")
	_elng = payload.get("end_longitude")
	if _elat is not None and str(_elat).strip() != "":
		v.end_latitude = float(_elat)
	if _elng is not None and str(_elng).strip() != "":
		v.end_longitude = float(_elng)
	if "checklist_answers" in payload:
		_ans = payload.get("checklist_answers")
		if _ans is not None and not isinstance(_ans, (dict, list)):
			raise ApiError("VALIDATION_ERROR", "checklist_answers must be object or list", http_status=400)
		v.checklist_answers = _ans
	if payload.get("shelf_photo_file_id"):
		v.shelf_photo_file_id = int(payload["shelf_photo_file_id"])
	van_lines = payload.get("van_sale_lines")
	if van_lines and outcome == "order":
		settings_v = get_or_create_distribution_settings(db, business_id)
		if getattr(settings_v, "enable_van_sales", False) and isinstance(van_lines, list) and van_lines:
			from app.services.distribution_phase3_service import get_van_for_user
			from app.services.distribution_documents import create_distribution_invoice
			from app.services.invoice_service import INVOICE_SALES
			from app.services.distribution_commercial_service import apply_promotions_to_lines
			from app.services.distribution_field_ops_service import apply_auto_promotions_if_needed, consume_van_lots_fefo

			promo_ids = None
			ex = payload.get("extra_info") if isinstance(payload.get("extra_info"), dict) else {}
			if isinstance(ex.get("promotion_ids"), list):
				promo_ids = [int(x) for x in ex["promotion_ids"]]
			promo_ids = apply_auto_promotions_if_needed(db, business_id, promo_ids)
			van_lines, _disc, applied = apply_promotions_to_lines(
				db, business_id, list(van_lines), promotion_ids=promo_ids, person_id=int(v.person_id),
			)
			if applied:
				_base = dict(v.extra_info or {})
				_base["promotion_ids"] = applied
				v.extra_info = _base

			van = get_van_for_user(db, business_id, user_id)
			if not van:
				raise ApiError("VALIDATION_ERROR", "No van assigned for van sale", http_status=400)
			if not v.document_id:
				# Fail-closed: بدون فاکتور موفق، ویزیت کامل نمی‌شود (بدون حوالهٔ خام انبار)
				inv = create_distribution_invoice(
					db,
					business_id,
					user_id,
					invoice_type=INVOICE_SALES,
					person_id=int(v.person_id),
					raw_lines=van_lines,
					warehouse_id=int(van.warehouse_id),
					description=f"فروش ون — ویزیت #{v.id}",
					meta={"distribution_visit_id": v.id, "van_id": van.id},
				)
				v.document_id = int(inv["id"])
				_base = dict(v.extra_info or {})
				_base["van_sale_invoice_id"] = int(inv["id"])
				_base["lines_count"] = len([ln for ln in van_lines if float(ln.get("quantity") or 0) > 0])
				lots_used = []
				for ln in van_lines:
					if ln.get("is_foc"):
						continue
					lots_used.extend(
						consume_van_lots_fefo(db, int(van.id), int(ln["product_id"]), float(ln["quantity"]))
					)
				if lots_used:
					_base["van_lots_consumed"] = lots_used
				v.extra_info = _base
	v.updated_at = datetime.utcnow()

	summary_parts = [f"ویزیت میدانی — نتیجه: {outcome}"]
	if v.notes:
		summary_parts.append(str(v.notes))
	summary = "\n".join(summary_parts)

	crm_activity_id = _create_visit_crm_activity(db, business_id, v, user_id, summary)
	if crm_activity_id:
		v.crm_activity_id = crm_activity_id

	db.commit()
	db.refresh(v)

	try:
		from app.services.workflow.workflow_trigger_service import trigger_distribution_visit_completed

		trigger_distribution_visit_completed(db, business_id, v.id, user_id)
	except Exception:
		import logging

		logging.getLogger(__name__).warning("distribution workflow trigger failed", exc_info=True)

	person = db.query(Person).filter(Person.id == v.person_id).first()
	pname = (person.alias_name or "").strip() if person else None
	return visit_to_dict(v, pname)


def _create_visit_crm_activity(db: Session, business_id: int, visit: DistributionFieldVisit, user_id: int, summary: str) -> Optional[int]:
	import logging

	logger = logging.getLogger(__name__)
	try:
		from adapters.db.models.crm import CrmActivity
		from app.services.document_numbering_service import generate_document_code
		from app.services.workflow.workflow_trigger_service import trigger_crm_activity_created

		code = generate_document_code(db, business_id, "crm_activity", date.today())
		act = CrmActivity(
			business_id=business_id,
			person_id=visit.person_id,
			lead_id=None,
			code=code,
			activity_type="note",
			subject="ویزیت میدانی / پخش مویرگی",
			description=summary[:8000] if summary else None,
			activity_date=datetime.utcnow(),
			deal_id=visit.deal_id,
			created_by_user_id=user_id,
			extra_info={"distribution_visit_id": visit.id},
		)
		db.add(act)
		db.flush()
		trigger_crm_activity_created(db, business_id, act.id, user_id)
		return int(act.id)
	except Exception as e:
		logger.warning("CRM activity for distribution visit skipped: %s", e)
		return None


def cancel_visit(
	db: Session,
	business_id: int,
	user_id: int,
	visit_id: int,
	reason: Optional[str],
	allow_manage_override: bool = False,
) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	v = db.query(DistributionFieldVisit).filter(DistributionFieldVisit.id == visit_id, DistributionFieldVisit.business_id == business_id).first()
	if not v:
		raise ApiError("NOT_FOUND", "Visit not found", http_status=404)
	if v.user_id != user_id and not allow_manage_override:
		raise ApiError("FORBIDDEN", "Only the visitor can cancel", http_status=403)
	if v.status != "in_progress":
		raise ApiError("VALIDATION_ERROR", "Visit is not in progress", http_status=400)
	v.status = "cancelled"
	v.ended_at = datetime.utcnow()
	v.outcome = "cancelled"
	v.no_order_reason = reason
	v.updated_at = datetime.utcnow()
	db.commit()
	db.refresh(v)
	person = db.query(Person).filter(Person.id == v.person_id).first()
	pname = (person.alias_name or "").strip() if person else None
	return visit_to_dict(v, pname)


def list_visits(
	db: Session,
	business_id: int,
	ctx: AuthContext,
	from_date: Optional[date],
	to_date: Optional[date],
	limit: int,
	skip: int,
) -> Tuple[List[Dict[str, Any]], int]:
	_ensure_plugin(db, business_id)
	q = db.query(DistributionFieldVisit).filter(DistributionFieldVisit.business_id == business_id)
	uid = _scope_visit_user_id(ctx, business_id)
	if uid is not None:
		q = q.filter(DistributionFieldVisit.user_id == uid)
	if from_date:
		q = q.filter(func.date(DistributionFieldVisit.started_at) >= from_date)
	if to_date:
		q = q.filter(func.date(DistributionFieldVisit.started_at) <= to_date)
	total = q.count()
	rows = q.order_by(DistributionFieldVisit.started_at.desc()).offset(skip).limit(limit).all()
	pids = [r.person_id for r in rows]
	pmap = {}
	if pids:
		for p in db.query(Person).filter(Person.id.in_(pids)).all():
			pmap[p.id] = (p.alias_name or "").strip()
	return [visit_to_dict(r, pmap.get(r.person_id)) for r in rows], total


def create_return_request(db: Session, business_id: int, user_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	person_id = int(payload.get("person_id") or 0)
	if person_id <= 0:
		raise ApiError("VALIDATION_ERROR", "person_id required", http_status=400)
	person = db.query(Person).filter(Person.id == person_id, Person.business_id == business_id).first()
	if not person:
		raise ApiError("NOT_FOUND", "Person not found", http_status=404)
	raw_lines = payload.get("lines") or []
	if not isinstance(raw_lines, list) or not raw_lines:
		raise ApiError("VALIDATION_ERROR", "lines must be a non-empty list", http_status=400)
	lines = _validate_return_lines(db, business_id, raw_lines)
	vid = payload.get("visit_id")
	source_document_id = payload.get("source_document_id")
	if source_document_id:
		_validate_document_for_business(db, business_id, int(source_document_id), person_id)
		# سقف مقدار نسبت به خطوط فاکتور منبع (در صورت وجود)
		_cap_return_lines_against_source(db, business_id, int(source_document_id), lines)
	# source_document_id را در meta خطوط نگه می‌داریم تا بدون مایگریشن ستون جدید حفظ شود
	meta_lines = list(lines)
	if source_document_id:
		meta_lines = [{**ln, "source_document_id": int(source_document_id)} for ln in lines]
	row = DistributionReturnRequest(
		business_id=business_id,
		person_id=person_id,
		visit_id=int(vid) if vid else None,
		status="pending",
		lines=meta_lines,
		notes=payload.get("notes"),
		created_by_user_id=user_id,
	)
	db.add(row)
	db.commit()
	db.refresh(row)
	return {
		"id": row.id,
		"person_id": row.person_id,
		"visit_id": row.visit_id,
		"status": row.status,
		"lines": row.lines,
		"source_document_id": int(source_document_id) if source_document_id else None,
		"notes": row.notes,
		"created_at": row.created_at.isoformat() if row.created_at else None,
	}


def list_return_requests(db: Session, business_id: int, ctx: AuthContext, status: Optional[str]) -> List[Dict[str, Any]]:
	_ensure_plugin(db, business_id)
	q = db.query(DistributionReturnRequest).filter(DistributionReturnRequest.business_id == business_id)
	uid = _scope_visit_user_id(ctx, business_id)
	if uid is not None:
		q = q.filter(DistributionReturnRequest.created_by_user_id == uid)
	if status:
		q = q.filter(DistributionReturnRequest.status == status)
	rows = q.order_by(DistributionReturnRequest.created_at.desc()).limit(500).all()
	from app.services.distribution_documents import person_label, user_label

	out = []
	for r in rows:
		out.append(
			{
				"id": r.id,
				"person_id": r.person_id,
				"person_name": person_label(db, r.person_id),
				"visit_id": r.visit_id,
				"status": r.status,
				"lines": r.lines,
				"notes": r.notes,
				"resolved_document_id": r.resolved_document_id,
				"created_by_user_id": r.created_by_user_id,
				"created_by_user_name": user_label(db, r.created_by_user_id),
				"created_at": r.created_at.isoformat() if r.created_at else None,
			}
		)
	return out


def resolve_return_request(
	db: Session,
	business_id: int,
	resolver_user_id: int,
	request_id: int,
	payload: Dict[str, Any],
) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	row = (
		db.query(DistributionReturnRequest)
		.filter(DistributionReturnRequest.id == request_id, DistributionReturnRequest.business_id == business_id)
		.first()
	)
	if not row:
		raise ApiError("NOT_FOUND", "Return request not found", http_status=404)
	if row.status != "pending":
		raise ApiError("VALIDATION_ERROR", "Return request already resolved", http_status=400)
	new_status = str(payload.get("status") or "").strip()
	if new_status not in ("approved", "rejected"):
		raise ApiError("VALIDATION_ERROR", "status must be approved or rejected", http_status=400)
	row.status = new_status
	row.resolved_by_user_id = resolver_user_id
	row.resolved_at = datetime.utcnow()
	warehouse_document_id: Optional[int] = None
	auto_invoice_id: Optional[int] = None
	if payload.get("resolved_document_id"):
		doc_id = int(payload["resolved_document_id"])
		_validate_document_for_business(db, business_id, doc_id, row.person_id)
		row.resolved_document_id = doc_id
	elif new_status == "approved":
		from app.services.distribution_documents import create_distribution_invoice
		from app.services.invoice_service import INVOICE_SALES_RETURN

		settings = get_or_create_distribution_settings(db, business_id)
		wh_id = getattr(settings, "default_source_warehouse_id", None)
		restock_to_van = bool(payload.get("restock_to_van"))
		if restock_to_van or (not wh_id and row.visit_id):
			from app.services.distribution_phase3_service import get_van_for_user

			visit = None
			if row.visit_id:
				visit = (
					db.query(DistributionFieldVisit)
					.filter(DistributionFieldVisit.id == row.visit_id, DistributionFieldVisit.business_id == business_id)
					.first()
				)
			if visit:
				van = get_van_for_user(db, business_id, int(visit.user_id))
				if van:
					wh_id = van.warehouse_id
		if not wh_id:
			from adapters.db.models.warehouse import Warehouse

			wh = (
				db.query(Warehouse)
				.filter(Warehouse.business_id == business_id)
				.order_by(Warehouse.is_default.desc(), Warehouse.id.asc())
				.first()
			)
			wh_id = wh.id if wh else None
		lines = row.lines if isinstance(row.lines, list) else []
		source_document_id = payload.get("source_document_id")
		if not source_document_id and lines:
			for ln in lines:
				if isinstance(ln, dict) and ln.get("source_document_id"):
					source_document_id = int(ln["source_document_id"])
					break
		if not wh_id:
			raise ApiError("VALIDATION_ERROR", "No warehouse available for return", http_status=400)
		if not lines:
			raise ApiError("VALIDATION_ERROR", "Return lines empty", http_status=400)
		# Fail-closed: تأیید مرجوعی بدون فاکتور برگشت مجاز نیست
		inv = create_distribution_invoice(
			db,
			business_id,
			resolver_user_id,
			invoice_type=INVOICE_SALES_RETURN,
			person_id=int(row.person_id),
			raw_lines=lines,
			warehouse_id=int(wh_id),
			description=f"برگشت از فروش — مرجوعی پخش #{row.id}",
			meta={
				"distribution_return_request_id": row.id,
				"source_document_id": int(source_document_id) if source_document_id else None,
			},
			source_document_id=int(source_document_id) if source_document_id else None,
		)
		auto_invoice_id = int(inv["id"])
		row.resolved_document_id = auto_invoice_id
		note_bits = []
		if row.notes:
			note_bits.append(str(row.notes))
		note_bits.append(f"[inv:{auto_invoice_id}]")
		if source_document_id:
			note_bits.append(f"[src:{int(source_document_id)}]")
		row.notes = "\n".join(note_bits) if note_bits else row.notes
	row.updated_at = datetime.utcnow()
	db.commit()
	db.refresh(row)
	return {
		"id": row.id,
		"status": row.status,
		"resolved_document_id": row.resolved_document_id,
		"warehouse_document_id": warehouse_document_id,
		"auto_invoice_id": auto_invoice_id,
		"resolved_at": row.resolved_at.isoformat() if row.resolved_at else None,
	}


def update_distribution_settings(db: Session, business_id: int, payload: Dict[str, Any]) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	row = get_or_create_distribution_settings(db, business_id)
	if "shared_routing_catalog" in payload:
		row.shared_routing_catalog = bool(payload["shared_routing_catalog"])
	if "require_visit_in_daily_plan" in payload:
		row.require_visit_in_daily_plan = bool(payload["require_visit_in_daily_plan"])
	if "geofence_radius_meters" in payload:
		row.geofence_radius_meters = max(0, int(payload["geofence_radius_meters"] or 0))
	if "require_geofence" in payload:
		row.require_geofence = bool(payload["require_geofence"])
	if "visit_checklist_template" in payload:
		tpl = payload["visit_checklist_template"]
		row.visit_checklist_template = tpl if isinstance(tpl, list) else []
	if "enable_van_sales" in payload:
		row.enable_van_sales = bool(payload["enable_van_sales"])
	if "default_source_warehouse_id" in payload:
		v = payload.get("default_source_warehouse_id")
		row.default_source_warehouse_id = int(v) if v else None
	if "enable_presell" in payload:
		row.enable_presell = bool(payload["enable_presell"])
	if "enable_promotions" in payload:
		row.enable_promotions = bool(payload["enable_promotions"])
	if "visitor_max_discount_percent" in payload:
		row.visitor_max_discount_percent = max(0.0, min(100.0, float(payload.get("visitor_max_discount_percent") or 0)))
	if "enable_suggested_order" in payload:
		row.enable_suggested_order = bool(payload["enable_suggested_order"])
	if "map_tile_source" in payload:
		from app.services.distribution_phase3_service import normalize_map_tile_source

		row.map_tile_source = normalize_map_tile_source(payload.get("map_tile_source"))
	if "memaps_api_key" in payload:
		raw_key = payload.get("memaps_api_key")
		key = str(raw_key).strip() if raw_key is not None else ""
		row.memaps_api_key = key[:255] if key else None
	if "share_live_location" in payload:
		row.share_live_location = bool(payload["share_live_location"])
	if "carry_over_missed_visits" in payload:
		row.carry_over_missed_visits = bool(payload["carry_over_missed_visits"])
	if "carry_over_days" in payload:
		row.carry_over_days = max(1, min(31, int(payload.get("carry_over_days") or 7)))
	if "require_pod_signature" in payload:
		row.require_pod_signature = bool(payload["require_pod_signature"])
	if "require_pod_photo" in payload:
		row.require_pod_photo = bool(payload["require_pod_photo"])
	if "nav_provider" in payload:
		nav = str(payload.get("nav_provider") or "neshan").strip().lower()
		row.nav_provider = nav if nav in ("neshan", "google", "waze") else "neshan"
	if "setup_completed" in payload:
		row.setup_completed = bool(payload["setup_completed"])
	if "auto_apply_promotions" in payload:
		row.auto_apply_promotions = bool(payload["auto_apply_promotions"])
	if "enable_perfect_store" in payload:
		row.enable_perfect_store = bool(payload["enable_perfect_store"])
	if "near_expiry_days" in payload:
		row.near_expiry_days = max(1, min(90, int(payload.get("near_expiry_days") or 14)))
	row.updated_at = datetime.utcnow()
	db.commit()
	db.refresh(row)
	return settings_to_dict(row)


def get_distribution_reports_dashboard(
	db: Session,
	business_id: int,
	ctx: AuthContext,
	from_date: date,
	to_date: date,
	target_user_id: Optional[int],
) -> Dict[str, Any]:
	_ensure_plugin(db, business_id)
	if target_user_id is not None and not _can_see_full_distribution_catalog(ctx, business_id):
		raise ApiError("FORBIDDEN", "Only managers can filter by visitor user", http_status=403)
	vq = db.query(DistributionFieldVisit).filter(DistributionFieldVisit.business_id == business_id)
	vq = vq.filter(func.date(DistributionFieldVisit.started_at) >= from_date)
	vq = vq.filter(func.date(DistributionFieldVisit.started_at) <= to_date)
	scope_uid = _scope_visit_user_id(ctx, business_id)
	if target_user_id is not None:
		vq = vq.filter(DistributionFieldVisit.user_id == int(target_user_id))
	elif scope_uid is not None:
		vq = vq.filter(DistributionFieldVisit.user_id == scope_uid)

	total_visits = vq.count()
	completed = vq.filter(DistributionFieldVisit.status == "completed").count()
	cancelled_vis = vq.filter(DistributionFieldVisit.status == "cancelled").count()
	inprog = vq.filter(DistributionFieldVisit.status == "in_progress").count()

	out_rows = (
		db.query(DistributionFieldVisit.outcome, func.count(DistributionFieldVisit.id))
		.filter(DistributionFieldVisit.business_id == business_id)
		.filter(func.date(DistributionFieldVisit.started_at) >= from_date)
		.filter(func.date(DistributionFieldVisit.started_at) <= to_date)
	)
	if target_user_id is not None:
		out_rows = out_rows.filter(DistributionFieldVisit.user_id == int(target_user_id))
	elif scope_uid is not None:
		out_rows = out_rows.filter(DistributionFieldVisit.user_id == scope_uid)
	out_rows = out_rows.group_by(DistributionFieldVisit.outcome).all()
	by_outcome: Dict[str, int] = {}
	for oc, cnt in out_rows:
		key = oc if oc else "unset"
		by_outcome[str(key)] = int(cnt)

	by_user: Optional[List[Dict[str, Any]]] = None
	from app.services.distribution_documents import document_net_amount, user_label

	sales_linked_total = 0.0
	sales_linked_count = 0
	completed_q = (
		db.query(DistributionFieldVisit)
		.filter(DistributionFieldVisit.business_id == business_id)
		.filter(func.date(DistributionFieldVisit.started_at) >= from_date)
		.filter(func.date(DistributionFieldVisit.started_at) <= to_date)
		.filter(DistributionFieldVisit.status == "completed")
	)
	if target_user_id is not None:
		completed_q = completed_q.filter(DistributionFieldVisit.user_id == int(target_user_id))
	elif scope_uid is not None:
		completed_q = completed_q.filter(DistributionFieldVisit.user_id == scope_uid)
	completed_list = completed_q.all()
	for visit in completed_list:
		if visit.document_id:
			amt = document_net_amount(db, int(visit.document_id))
			if amt is not None:
				sales_linked_total += float(amt)
				sales_linked_count += 1

	plan_stops = 0
	visited_persons: set[int] = {int(v.person_id) for v in completed_list}
	if target_user_id is not None or scope_uid is not None:
		uid_cov = int(target_user_id) if target_user_id is not None else int(scope_uid)  # type: ignore[arg-type]
		from datetime import timedelta

		span_days = (to_date - from_date).days
		if span_days <= 62:
			d = from_date
			while d <= to_date:
				plan = get_daily_plan(db, business_id, uid_cov, d)
				plan_stops += len(plan.get("items") or [])
				d = d + timedelta(days=1)
	coverage_pct = None
	if plan_stops > 0:
		coverage_pct = round(100.0 * min(len(visited_persons), plan_stops) / float(plan_stops), 1)

	if _can_see_full_distribution_catalog(ctx, business_id):
		uq = (
			db.query(DistributionFieldVisit.user_id, func.count(DistributionFieldVisit.id))
			.filter(DistributionFieldVisit.business_id == business_id)
			.filter(func.date(DistributionFieldVisit.started_at) >= from_date)
			.filter(func.date(DistributionFieldVisit.started_at) <= to_date)
		)
		if target_user_id is not None:
			uq = uq.filter(DistributionFieldVisit.user_id == int(target_user_id))
		uq = uq.group_by(DistributionFieldVisit.user_id).all()
		by_user = [
			{"user_id": int(uid), "user_name": user_label(db, int(uid)), "visit_count": int(c)}
			for uid, c in uq
		]

	rq = db.query(DistributionReturnRequest).filter(DistributionReturnRequest.business_id == business_id)
	rq = rq.filter(func.date(DistributionReturnRequest.created_at) >= from_date)
	rq = rq.filter(func.date(DistributionReturnRequest.created_at) <= to_date)
	if target_user_id is not None:
		rq = rq.filter(DistributionReturnRequest.created_by_user_id == int(target_user_id))
	elif scope_uid is not None:
		rq = rq.filter(DistributionReturnRequest.created_by_user_id == scope_uid)

	def _cnt(st: str) -> int:
		return int(rq.filter(DistributionReturnRequest.status == st).count())

	# تسویه‌های بازه
	from adapters.db.models.distribution import DistributionDailySettlement

	sq = db.query(DistributionDailySettlement).filter(DistributionDailySettlement.business_id == business_id)
	sq = sq.filter(DistributionDailySettlement.settlement_date >= from_date)
	sq = sq.filter(DistributionDailySettlement.settlement_date <= to_date)
	if target_user_id is not None:
		sq = sq.filter(DistributionDailySettlement.user_id == int(target_user_id))
	elif scope_uid is not None:
		sq = sq.filter(DistributionDailySettlement.user_id == scope_uid)
	settlements = sq.all()
	sett_confirmed = [s for s in settlements if s.status == "confirmed"]
	sett_draft = [s for s in settlements if s.status == "draft"]
	variance_abs = sum(abs(float(s.variance or 0)) for s in sett_confirmed)
	collected_total = sum(
		float(s.cash_collected or 0)
		+ float(s.cheque_collected or 0)
		+ float(s.card_collected or 0)
		+ float(s.other_collected or 0)
		for s in sett_confirmed
	)

	# sell-through تقریبی: فروش لینک‌شده / (فروش لینک‌شده + موجودی نمونه نه) — فقط نسبت سفارش به تکمیل
	orders = int(by_outcome.get("order") or 0)
	sell_through_pct = round(100.0 * orders / completed, 1) if completed > 0 else None

	return {
		"from_date": from_date.isoformat(),
		"to_date": to_date.isoformat(),
		"target_user_id": target_user_id,
		"visits": {
			"total_records": total_visits,
			"completed": completed,
			"cancelled": cancelled_vis,
			"in_progress": inprog,
			"by_outcome": by_outcome,
			"sales_linked_count": sales_linked_count,
			"sales_linked_net_total": sales_linked_total,
			"plan_stops_in_range": plan_stops,
			"unique_persons_visited": len(visited_persons),
			"coverage_percent": coverage_pct,
			"order_rate_percent": sell_through_pct,
		},
		"by_user": by_user,
		"returns": {
			"pending": _cnt("pending"),
			"approved": _cnt("approved"),
			"rejected": _cnt("rejected"),
		},
		"settlements": {
			"draft_count": len(sett_draft),
			"confirmed_count": len(sett_confirmed),
			"confirmed_collected_total": collected_total,
			"confirmed_variance_abs_total": variance_abs,
		},
	}