133 lines
4.7 KiB
Python
133 lines
4.7 KiB
Python
from math import asin, cos, radians, sin, sqrt
|
|
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session, selectinload
|
|
|
|
from app.models import AttendancePoint, PersonAttendancePoint, Personnel, Role
|
|
|
|
|
|
DEFAULT_ATTENDANCE_POINT_NAME = "宁波嘉恒智能科技有限公司"
|
|
EARTH_RADIUS_METERS = 6371000
|
|
|
|
|
|
def clean_point_name(value: str | None) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def ensure_default_attendance_point(db: Session) -> AttendancePoint:
|
|
point = db.get(AttendancePoint, DEFAULT_ATTENDANCE_POINT_NAME)
|
|
if point is not None:
|
|
return point
|
|
point = AttendancePoint(
|
|
name=DEFAULT_ATTENDANCE_POINT_NAME,
|
|
radius_meters=500,
|
|
is_active=True,
|
|
)
|
|
db.add(point)
|
|
db.flush()
|
|
return point
|
|
|
|
|
|
def attendance_point_names(db: Session) -> list[str]:
|
|
names = list(db.scalars(select(AttendancePoint.name).where(AttendancePoint.is_active.is_(True)).order_by(AttendancePoint.name)))
|
|
if not names:
|
|
point = ensure_default_attendance_point(db)
|
|
db.commit()
|
|
names = [point.name]
|
|
return names
|
|
|
|
|
|
def accessible_point_names(db: Session, user: Personnel) -> list[str]:
|
|
if user.role == Role.manager:
|
|
return attendance_point_names(db)
|
|
|
|
rows = list(
|
|
db.scalars(
|
|
select(PersonAttendancePoint.attendance_point_name)
|
|
.join(AttendancePoint, AttendancePoint.name == PersonAttendancePoint.attendance_point_name)
|
|
.where(
|
|
PersonAttendancePoint.phone == user.phone,
|
|
AttendancePoint.is_active.is_(True),
|
|
)
|
|
.order_by(PersonAttendancePoint.attendance_point_name)
|
|
)
|
|
)
|
|
return rows
|
|
|
|
|
|
def accessible_points(db: Session, user: Personnel) -> list[AttendancePoint]:
|
|
names = accessible_point_names(db, user)
|
|
return list(
|
|
db.scalars(
|
|
select(AttendancePoint)
|
|
.where(AttendancePoint.name.in_(names), AttendancePoint.is_active.is_(True))
|
|
.order_by(AttendancePoint.name)
|
|
)
|
|
)
|
|
|
|
|
|
def require_attendance_point_access(db: Session, user: Personnel, point_name: str | None) -> str:
|
|
name = clean_point_name(point_name) or DEFAULT_ATTENDANCE_POINT_NAME
|
|
if db.get(AttendancePoint, name) is None:
|
|
raise ValueError(f"没有该考勤点：{name}")
|
|
if user.role != Role.manager and name not in accessible_point_names(db, user):
|
|
raise PermissionError(f"无该考勤点权限：{name}")
|
|
return name
|
|
|
|
|
|
def distance_meters(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
|
|
delta_lat = radians(lat2 - lat1)
|
|
delta_lng = radians(lng2 - lng1)
|
|
first_lat = radians(lat1)
|
|
second_lat = radians(lat2)
|
|
value = (
|
|
sin(delta_lat / 2) * sin(delta_lat / 2)
|
|
+ cos(first_lat) * cos(second_lat) * sin(delta_lng / 2) * sin(delta_lng / 2)
|
|
)
|
|
return 2 * EARTH_RADIUS_METERS * asin(sqrt(min(1, value)))
|
|
|
|
|
|
def validate_attendance_point_location(
|
|
db: Session,
|
|
point_name: str,
|
|
latitude: float | None,
|
|
longitude: float | None,
|
|
*,
|
|
action_name: str = "扫码报工",
|
|
) -> None:
|
|
if latitude is None or longitude is None:
|
|
raise ValueError(f"{action_name}需要校验当前位置")
|
|
point = db.get(AttendancePoint, clean_point_name(point_name))
|
|
if point is None or not point.is_active:
|
|
raise ValueError(f"没有该考勤点：{point_name}")
|
|
if point.latitude is None or point.longitude is None:
|
|
raise ValueError(f"考勤点【{point.name}】未配置经纬度")
|
|
distance = distance_meters(float(latitude), float(longitude), float(point.latitude), float(point.longitude))
|
|
radius = int(point.radius_meters or 500)
|
|
if distance > radius:
|
|
raise PermissionError(
|
|
f"当前位置距离考勤点【{point.name}】约{round(distance)}米，超过{radius}米范围，不能{action_name}"
|
|
)
|
|
|
|
|
|
def person_with_access_points(db: Session, phone: str) -> Personnel | None:
|
|
return db.scalar(
|
|
select(Personnel)
|
|
.options(selectinload(Personnel.roles), selectinload(Personnel.attendance_points))
|
|
.where(Personnel.phone == phone)
|
|
)
|
|
|
|
|
|
def set_person_attendance_points(db: Session, phone: str, point_names: list[str]) -> None:
|
|
cleaned = [clean_point_name(name) for name in point_names if clean_point_name(name)]
|
|
if not cleaned:
|
|
cleaned = [ensure_default_attendance_point(db).name]
|
|
valid_names = set(attendance_point_names(db))
|
|
missing = [name for name in cleaned if name not in valid_names]
|
|
if missing:
|
|
raise ValueError(f"没有该考勤点：{'、'.join(missing)}")
|
|
db.query(PersonAttendancePoint).filter(PersonAttendancePoint.phone == phone).delete()
|
|
db.flush()
|
|
for name in dict.fromkeys(cleaned):
|
|
db.add(PersonAttendancePoint(phone=phone, attendance_point_name=name))
|