devops_admin
/
odex25_standard
mirror of https://github.com/hydracp9/odex25_standard
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

[IMP] odex_mobile: add push notifications mixin

This commit is contained in:
Abdurrahman Saber 2025-09-20 01:24:37 +03:00
parent 6d53a4d550
commit 1b8f53c192
5 changed files with 86 additions and 102 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
odex25_mobile/odex_mobile/controllers/attendance.py
View File

@ -374,11 +374,7 @@ class AttendanceController(http.Controller):
partner_id = emp.user_id.partner_id
partner_id.send_notification(
subject, msg, data=None, all_device=True)
data = {
'title':subject,
'body':msg,
}
emp.user_push_notification(data)
emp.send_push_notification(title=subject, body=msg)
def create_log(self, employee, longitude, latitude):
attendance = http.request.env['attendance.log'].create({

1
odex25_mobile/odex_mobile/models/__init__.py
View File

@ -10,3 +10,4 @@ from . import hr_firebase_notification
from . import base
from . import ir_property
from . import resource
from . import push_notification_mixin

64
odex25_mobile/odex_mobile/models/hr_employee.py
View File

@ -4,79 +4,19 @@ from odoo import models,fields,api,_
from odoo.exceptions import ValidationError
import random
import json
import json, requests
import json, requests, base64
import google.auth.transport.requests
from google.oauth2 import service_account
BASE_URL = 'https://fcm.googleapis.com'
SCOPES = ['https://www.googleapis.com/auth/firebase.messaging']
import tempfile
import os
class HrEmployee(models.Model):
_inherit = 'hr.employee'
_name = 'hr.employee'
_inherit = ['hr.employee', 'push.notification.mixin']
device_id = fields.Char(string="Employee Device ")
fcm_token = fields.Char(string='FCM Token')
attendance_log_ids = fields.One2many('attendance.log','employee_id',string="Attendance Log")
message_sent = fields.Boolean(string="Message Sent", default=False)
# Used internally by the API
last_active_time = fields.Datetime(readonly=True)
attendance_zone_id = fields.Many2many('attendance.zone', string='Attendance Zone')
def user_push_notification(self, notification):
def _get_access_token(json_file):
"""Retrieve a valid access token that can be used to authorize requests.
:return: Access token.
"""
credentials = service_account.Credentials.from_service_account_file(
json_file, scopes=SCOPES)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
return credentials.token
try:
json_file_name = 'service-account'
base64_service_account = base64.decodebytes(self.env.user.company_id.service_account)
service_account_json = json.loads(base64_service_account)
with tempfile.TemporaryDirectory() as temp_dir:
temp_file_path = os.path.join(temp_dir, json_file_name)
with open(temp_file_path, 'w') as temp_file:
temp_file.write(base64_service_account.decode('UTF-8'))
header = {
'Content-Type': 'application/json',
'Authorization': 'Bearer %s' % (_get_access_token(temp_file_path))
}
body = json.dumps({
"message": {
"token": self.fcm_token,
"notification": notification,
"android": {
"notification": {
"sound": "default"
}
},
"apns": {
"payload": {
"aps": {
"sound": "default"
}
}
}
}
})
FCM_ENDPOINT = 'v1/projects/' + service_account_json['project_id'] + '/messages:send'
FCM_URL = BASE_URL + '/' + FCM_ENDPOINT
respons = requests.post(url=FCM_URL, data=body, headers=header)
return True
except Exception as e:
return False
def create_employee_notification(self):
emp_notif_obj = self.env['employee.notification']
holiday_obj = self.env['hr.holidays']

78
odex25_mobile/odex_mobile/models/push_notification_mixin.py Normal file
View File

@ -0,0 +1,78 @@
import json, requests, base64
import tempfile
import os
import logging
from odoo import models, fields, api, _
from odoo.exceptions import UserError
import google.auth.transport.requests
from google.oauth2 import service_account
BASE_URL = 'https://fcm.googleapis.com'
SCOPES = ['https://www.googleapis.com/auth/firebase.messaging']
_logger = logging.getLogger(__name__)
class PushNotificationMixin(models.AbstractModel):
_name = 'push.notification.mixin'
_description = 'Push Notification Mixin'
fcm_token = fields.Char(string='FCM Token')
def send_push_notification(self, title, body):
def _get_access_token(json_file):
"""Retrieve a valid access token that can be used to authorize requests.
:return: Access token.
"""
credentials = service_account.Credentials.from_service_account_file(
json_file, scopes=SCOPES)
request = google.auth.transport.requests.Request()
credentials.refresh(request)
return credentials.token
for rec in self:
if not rec.fcm_token:
raise UserError(_("FCM Token is required"))
if not self.env.user.company_id.service_account:
raise UserError(_("Firebase Service Account file is required"))
try:
json_file_name = 'service-account'
base64_service_account = base64.decodebytes(self.env.user.company_id.service_account)
service_account_json = json.loads(base64_service_account)
with tempfile.TemporaryDirectory() as temp_dir:
temp_file_path = os.path.join(temp_dir, json_file_name)
with open(temp_file_path, 'w') as temp_file:
temp_file.write(base64_service_account.decode('UTF-8'))
header = {
'Content-Type': 'application/json',
'Authorization': 'Bearer %s' % (_get_access_token(temp_file_path))
}
body = json.dumps({
"message": {
"token": rec.fcm_token,
"notification": {
"title": title,
"body": body
},
"android": {
"notification": {
"sound": "default"
}
},
"apns": {
"payload": {
"aps": {
"sound": "default"
}
}
}
}
})
FCM_ENDPOINT = 'v1/projects/' + service_account_json['project_id'] + '/messages:send'
FCM_URL = BASE_URL + '/' + FCM_ENDPOINT
requests.post(url=FCM_URL, data=body, headers=header)
except Exception as e:
_logger.error(f"Failed to send push notification for {rec.name}: {e}")

37
odex25_mobile/odex_mobile/models/res_partner.py
View File

@ -6,22 +6,13 @@ from odoo import models, fields, api
_logger = logging.getLogger(__name__)
class Partner(models.Model):
_inherit = 'res.partner'
_name = 'res.partner'
_inherit = ['res.partner', 'push.notification.mixin']
firebase_registration_ids = fields.One2many(
"firebase.registration", "partner_id", readonly=True
)
def send_msg(self, partner_ids, msg, subject):
emp = self.env['hr.employee'].sudo().search([('user_id', 'in', partner_ids.user_ids.ids)])
if emp.user_id.partner_id:
partner_id = emp.user_id.partner_id
# partner_id.send_notification(subject, msg, data=None, all_device=True)
data = {
'title':subject,
'body':msg,
}
emp.user_push_notification(data)
def send_notification(self, message_title, message_body, data=None, all_device=True):
notification_data = {
"title": str(message_title),
@ -34,7 +25,7 @@ class Partner(models.Model):
notification = self.env['firebase.notification'].sudo().create(notification_data)
if all_device:
self.send_msg(notification.partner_ids,str(message_title),str(message_body))
notification.partner_ids.send_push_notification(str(message_title), str(message_body))
for reg in self.firebase_registration_ids:
reg.with_context(lang=self.lang).send_message(
message_title,
@ -57,25 +48,3 @@ class Partner(models.Model):
message_body,
data=data
)
def user_push_notification(self, fcm_token):
url = "https://fcm.googleapis.com/fcm/send"
headers = {
'Content-Type': 'application/json',
'Authorization': f'key={self.env.user.company_id.fcm_server_key}'
}
body = json.dumps({
"to": fcm_token,
"direct_boot_ok": True,
"notification": {
"title": "Test",
"body": "test"
}
})
try:
response = requests.post(url=url, data=body, headers=headers)
response.raise_for_status()
return True
except requests.exceptions.RequestException as e:
_logger.error(f"Failed to send push notification: {e}")
return False