from django.shortcuts import get_object_or_404
from api.models import User
from devices.models import Device
from api.notifications import send_sms
import requests
from apibase.env import env, BASE_DIR
import os
import logging
from celery import shared_task
from django.utils import timezone
from api.notifications import send_telegram_markdown
logger = logging.getLogger(__name__)
env.read_env(os.path.join(BASE_DIR, ".env"))
PERSON_VERIFY_BASE_URL = env.str("PERSON_VERIFY_BASE_URL", default="") # type: ignore
OMADA_PROXY_API_KEY = env.str("OMADA_PROXY_API_KEY", default="") # type: ignore
OMADA_PROXY_URL = env("OMADA_PROXY_URL", default="") # type: ignore
OMADA_SITE_ID = env("OMADA_SITE_ID", default="") # type: ignore
OMADA_GROUP_ID = env("OMADA_GROUP_ID", default="") # type: ignore
@shared_task
def add(x, y):
print(f"Adding {x} and {y}")
return x + y
@shared_task
def deactivate_expired_devices():
expired_devices = Device.objects.filter(
expiry_date__lte=timezone.localtime(timezone.now()), is_active=True
).select_related("user")
print("Expired Devices: ", expired_devices)
count = expired_devices.count()
user_devices_map = {}
for device in expired_devices:
if device.user and device.user.mobile:
if device.user.mobile not in user_devices_map:
user_devices_map[device.user.mobile] = []
user_devices_map[device.user.mobile].append(device.name)
for mobile, device_names in user_devices_map.items():
if not mobile:
continue
device_list = "\n".join(
[f"{i + 1}. {name}" for i, name in enumerate(device_names)]
)
print("device list: ", device_list)
send_sms(
mobile,
f"Dear {mobile}, \n\nThe following devices have expired: \n{device_list}. \n\nPlease make a payment to keep your devices active. \n\n- SAR Link",
)
# expired_devices.update(is_active=False)
print(f"Total {count} expired devices.")
return {
"total_expired_devices": count,
}
def get_existing_omada_devices():
"""
Get existing Omada devices from the database.
:return: List of existing device names.
"""
if not OMADA_PROXY_URL:
raise ValueError(
"OMADA_PROXY_URL is not set. Please set it in your environment variables."
)
try:
response = requests.get(
f"{OMADA_PROXY_URL}/9fd0cffa3475a74ae4e4d37de0d12414/api/v2/sites/66dcddb804aa0d2978cf145f/setting/profiles/groups",
headers={"X-API-Key": str(OMADA_PROXY_API_KEY)},
)
print("Response: ", response.status_code)
data = response.json()
existing_devices = []
if "result" in data and len(data["result"]["data"]) > 0:
last_entry = data["result"]["data"][-1]
print("Last Entry: ", last_entry)
if "macAddressList" in last_entry:
existing_devices = last_entry["macAddressList"]
print(existing_devices)
return existing_devices
except requests.RequestException as e:
print(f"Error fetching existing devices: {e}")
return []
@shared_task
def add_new_devices_to_omada(new_devices: list[dict]):
"""
Add new devices to Omada.
:param new_devices: List of new device names to add.
"""
if not OMADA_SITE_ID:
raise ValueError(
"OMADA_SITE_ID is not set. Please set it in your environment variables."
)
if not OMADA_GROUP_ID:
raise ValueError(
"OMADA_GROUP_ID is not set. Please set it in your environment variables."
)
if not OMADA_PROXY_URL:
raise ValueError(
"OMADA_PROXY_URL is not set. Please set it in your environment variables."
)
try:
PAYLOAD = {
"name": "REGISTERED_DEVICES",
"type": 2,
"resource": 0,
"ipList": None,
"ipv6List": None,
"macAddressList": None,
"portList": None,
"countryList": None,
"portType": None,
"portMaskList": None,
"domainNamePort": None,
}
existing_devices = get_existing_omada_devices()
PAYLOAD["macAddressList"] = existing_devices
print("Payload with existing devices: ", PAYLOAD)
for device in new_devices:
print("Device in loop: ", device)
PAYLOAD["macAddressList"].append(
{
"macAddress": device["mac"],
"name": device["name"],
}
)
print("New Payload: ", PAYLOAD)
print(
f"{OMADA_PROXY_URL}/9fd0cffa3475a74ae4e4d37de0d12414/api/v2/sites/{OMADA_SITE_ID}/setting/profiles/groups/2/{OMADA_GROUP_ID}"
)
response = requests.patch(
f"{OMADA_PROXY_URL}/9fd0cffa3475a74ae4e4d37de0d12414/api/v2/sites/{OMADA_SITE_ID}/setting/profiles/groups/2/{OMADA_GROUP_ID}",
headers={"X-API-Key": str(OMADA_PROXY_API_KEY)},
json=PAYLOAD,
)
print("Response: ", response.status_code)
if response.status_code == 200:
print("Devices successfully added.")
print(response.json())
else:
print(f"Failed to add devices: {response.text}")
except requests.RequestException as e:
print(f"Error adding devices: {e}")
def verify_user_with_person_api_task(user_id: int):
"""
Verify the user with the Person API.
:param user_id: The ID of the user to verify.
"""
user = get_object_or_404(User, id=user_id)
# Call the Person API to verify the user
verification_failed_message = f"""
_The following user verification failed_:
*ID Card:* {user.id_card} \n
*Name:* {user.first_name} {user.last_name} \n
*House Name:* {user.address} \n
*Date of Birth:* {user.dob} \n
*Island:* {(user.atoll.name if user.atoll else "N/A")} {(user.island.name if user.island else "N/A")} \n
*Mobile:* {user.mobile} \n
Visit [SAR Link Portal](https://portal.sarlink.net) to manually verify this user.
"""
if not PERSON_VERIFY_BASE_URL:
raise ValueError(
"PERSON_VERIFY_BASE_URL is not set in the environment variables."
)
response = requests.get(f"{PERSON_VERIFY_BASE_URL}/api/person/{user.id_card}")
if response.status_code == 200:
data = response.json()
api_nic = data.get("nic")
api_name = data.get("name_en")
api_house_name = data.get("house_name_en")
api_dob = data.get("dob")
api_atoll = data.get("atoll_en")
api_island_name = data.get("island_name_en")
if not user.mobile or user.dob is None:
logger.error("User mobile or date of birth is not set.")
return None
if not user.island or user.atoll is None:
logger.error("User island or atoll is not set.")
return None
logger.info(f"API nic: {api_nic}")
logger.info(f"API name: {api_name}")
logger.info(f"API house name: {api_house_name}")
logger.info(f"API dob: {api_dob}")
logger.info(f"API atoll: {api_atoll}")
logger.info(f"API island name: {api_island_name}")
user_nic = user.id_card
user_name = f"{user.first_name} {user.last_name}"
user_house_name = user.address
user_dob = user.dob.isoformat()
logger.info(f"User nic: {user_nic}")
logger.info(f"User name: {user_name}")
logger.info(f"User house name: {user_house_name}")
logger.info(f"User dob: {user_dob}")
logger.info(f"User atoll: {user.atoll}")
logger.info(f"User island name: {user.island}")
logger.info(
f"case User atoll: {user.atoll.name == api_atoll.strip() if api_atoll else False}"
) # Defensive check for api_atoll
logger.info(f"api atoll type: {type(api_atoll)}")
logger.info(f"user atoll type: {type(user.atoll.name)}")
logger.info(
f"case User island name: {user.island.name == api_island_name.strip() if api_island_name else False}"
) # Defensive check for api_island_name
logger.info(f"api island name type: {type(api_island_name)}")
logger.info(f"user island name type: {type(user.island.name)}")
if (
data.get("nic") == user.id_card
and data.get("name_en") == f"{user.first_name} {user.last_name}"
and data.get("house_name_en") == user.address
and data.get("dob").split("T")[0] == user.dob.isoformat()
and data.get("atoll_en").strip() == user.atoll.name
and data.get("island_name_en").strip() == user.island.name
):
user.verified = True
user.save()
send_sms(
user.mobile,
f"Dear {user.first_name} {user.last_name}, \n\nYour account has been successfully and verified. \n\nYou can now manage your devices and make payments through our portal at https://portal.sarlink.net. \n\n - SAR Link",
)
return True
else:
user.verified = False
user.save()
send_sms(
user.mobile,
f"Dear {user.first_name} {user.last_name}, \n\nYour account registration is being processed. \n\nWe will notify you once verification is complete. \n\n - SAR Link",
)
send_telegram_markdown(message=verification_failed_message)
return False
else:
# Handle the error case
print(f"Error verifying user: {response.status_code} - {response.text}")
return False