mirror of
https://github.com/itsdave-de/fusionpbx_connect.git
synced 2025-05-06 23:55:21 +02:00
108 lines
4.3 KiB
Python
108 lines
4.3 KiB
Python
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
from frappe.utils.password import get_decrypted_password
|
|
from frappe.utils import add_to_date, get_datetime
|
|
import frappe
|
|
import requests
|
|
|
|
def traccar_auth(ts):
|
|
"""authenticates on traccar server and returns cookies"""
|
|
response = requests.get(
|
|
f"{ts.traccar_server}/api/session",
|
|
params = {
|
|
'token': get_decrypted_password('Traccar Settings', ts.name, 'traccar_token')
|
|
}
|
|
)
|
|
|
|
if response.status_code == 200:
|
|
return response.cookies
|
|
else:
|
|
frappe.throw(f"Authentication failed: {response.status_code}: {response.reason}<br />Dict ts: {ts.as_dict()}")
|
|
|
|
def get_devices():
|
|
ts = frappe.get_doc('Traccar Settings')
|
|
|
|
try:
|
|
devices_response = requests.get(
|
|
f"{ts.traccar_server}/api/devices",
|
|
cookies=traccar_auth(ts)
|
|
)
|
|
devices_response.raise_for_status()
|
|
|
|
except requests.HTTPError as e:
|
|
frappe.throw(f"HTTP error occurred: {e}")
|
|
except Exception as e:
|
|
frappe.throw(f"Error fetching devices: {e}")
|
|
|
|
return devices_response.json()
|
|
|
|
|
|
@frappe.whitelist()
|
|
def get_trips_for_device():
|
|
ts = frappe.get_doc('Traccar Settings')
|
|
frappe.logger().info("Starting import Trips from Traccar server ...")
|
|
for dev in get_devices():
|
|
try:
|
|
trips_response = requests.get(
|
|
f"{ts.traccar_server}/api/reports/trips",
|
|
cookies=traccar_auth(ts),
|
|
params={
|
|
'deviceId': dev['id'],
|
|
'from': (datetime.now() - timedelta(days=1)).strftime('%Y-%m-%dT23:00:00.000Z'),
|
|
'to': datetime.now().strftime('%Y-%m-%dT22:59:59.999Z')
|
|
}
|
|
)
|
|
trips_response.raise_for_status()
|
|
except requests.HTTPError as e:
|
|
frappe.throw(f"HTTP error occurred: {e}")
|
|
except Exception as e:
|
|
frappe.throw(f"Error fetching trips: {e}")
|
|
|
|
#if len(trips_response.json()) > 0:
|
|
# for debug in trips_response.json():
|
|
# print(f"Distance: {debug['distance']} | Average Speed: {debug['averageSpeed']}")
|
|
|
|
for trip in trips_response.json():
|
|
# Assuming 'device_id' and 'startTime' can uniquely identify a trip
|
|
unique_id = f"{trip['deviceId']}-{trip['startTime']}"
|
|
|
|
# Check if the document already exists
|
|
existing_trip = frappe.db.exists('Trip', {'unique_id': unique_id})
|
|
|
|
if existing_trip:
|
|
trip_doc = frappe.get_doc('Trip', existing_trip)
|
|
else:
|
|
trip_doc = frappe.new_doc('Trip')
|
|
trip_doc.unique_id = unique_id # Atribui o identificador único
|
|
|
|
# Atualiza ou define os campos
|
|
trip_doc.device_id = trip['deviceId']
|
|
trip_doc.start = datetime.fromisoformat(trip['startTime']).replace(
|
|
tzinfo=ZoneInfo("UTC")).astimezone(ZoneInfo("Europe/Berlin")
|
|
).strftime('%Y-%m-%d %H:%M:%S')
|
|
trip_doc.end = datetime.fromisoformat(trip['endTime']).replace(
|
|
tzinfo=ZoneInfo("UTC")).astimezone(ZoneInfo("Europe/Berlin")
|
|
).strftime('%Y-%m-%d %H:%M:%S')
|
|
trip_doc.distance = trip['distance']
|
|
trip_doc.average_speed = trip['averageSpeed']
|
|
trip_doc.max_speed = trip['maxSpeed']
|
|
trip_doc.spent_fuel = trip['spentFuel']
|
|
trip_doc.start_position_id = trip['startPositionId']
|
|
trip_doc.end_position_id = trip['endPositionId']
|
|
trip_doc.start_lat = trip['startLat']
|
|
trip_doc.start_lon = trip['startLon']
|
|
trip_doc.end_lat = trip['endLat']
|
|
trip_doc.end_lon = trip['endLon']
|
|
trip_doc.start_address = trip['startAddress']
|
|
trip_doc.end_address = trip['endAddress']
|
|
trip_doc.duration = trip['duration']
|
|
trip_doc.driver_unique_id = trip['driverUniqueId']
|
|
trip_doc.tracker = frappe.get_value('Tracker', {'portal_id': trip['deviceId']}, 'portal_name')
|
|
trip_doc.fleet_vehicle = frappe.get_value('Fleet Vehicle', {'tracker': trip_doc.tracker}, 'name')
|
|
|
|
trip_doc.save(ignore_permissions=True)
|
|
|
|
frappe.db.commit()
|
|
|
|
frappe.logger().info("End of task of import")
|