408 lines
18 KiB
Python
408 lines
18 KiB
Python
# -*- coding: utf-8 -*-
|
|
from odoo import http
|
|
from odoo.http import request
|
|
|
|
def normalize_phone_search(phone):
|
|
if not phone:
|
|
return []
|
|
digits = ''.join(c for c in phone if c.isdigit())
|
|
if not digits:
|
|
return []
|
|
candidates = [digits]
|
|
if digits.startswith('62'):
|
|
candidates.append('0' + digits[2:])
|
|
candidates.append(digits[2:])
|
|
elif digits.startswith('0'):
|
|
candidates.append('62' + digits[1:])
|
|
candidates.append(digits[1:])
|
|
else:
|
|
candidates.append('0' + digits)
|
|
candidates.append('62' + digits)
|
|
return list(set(candidates))
|
|
|
|
class AppNotificationController(http.Controller):
|
|
|
|
@http.route('/api/loyalty/fetch_notifications', type='jsonrpc', auth='user', methods=['POST'], csrf=False)
|
|
def fetch_notifications(self, **kw):
|
|
"""
|
|
Endpoint for the Flutter app Background Task and In-App notification center.
|
|
Returns image_128 (base64 thumbnail) for list display.
|
|
The Flutter detail screen loads the full image via /web/image/ with session cookie.
|
|
"""
|
|
import base64 as b64mod
|
|
user = request.env.user
|
|
partner = user.partner_id
|
|
|
|
last_id = kw.get('last_id', 0)
|
|
|
|
# Give them global notifications OR notifications tagged explicitly to them
|
|
domain = [
|
|
('id', '>', last_id),
|
|
'|',
|
|
('partner_ids', 'in', [partner.id]),
|
|
('is_global', '=', True)
|
|
]
|
|
|
|
notifications = request.env['mapan.app.notification'].sudo().search_read(
|
|
domain,
|
|
['id', 'title', 'body', 'create_date', 'image_128'],
|
|
order='create_date desc',
|
|
limit=50
|
|
)
|
|
|
|
# Normalize image_128 bytes → base64 string for JSON transport
|
|
for notif in notifications:
|
|
img = notif.get('image_128')
|
|
if img and isinstance(img, bytes):
|
|
notif['image_128'] = img.decode('utf-8')
|
|
elif not img:
|
|
notif['image_128'] = None
|
|
notif['has_image'] = bool(notif['image_128'])
|
|
|
|
return {
|
|
'status': 'success',
|
|
'data': notifications
|
|
}
|
|
|
|
@http.route('/api/loyalty/branches', type='jsonrpc', auth='public', methods=['POST'], csrf=False)
|
|
def fetch_branches(self, **kw):
|
|
"""
|
|
Public endpoint for the Flutter app to get branches without exposing API keys.
|
|
"""
|
|
try:
|
|
branches = request.env['res.company'].sudo().search_read(
|
|
[('parent_id', '!=', False)],
|
|
['name', 'street', 'city', 'phone'],
|
|
limit=50
|
|
)
|
|
return {'status': 'success', 'data': branches}
|
|
except Exception as e:
|
|
return {'status': 'error', 'message': str(e)}
|
|
|
|
@http.route('/api/loyalty/send_otp', type='jsonrpc', auth='public', methods=['POST'], csrf=False)
|
|
def send_otp(self, **kw):
|
|
"""
|
|
Public endpoint to send a verification OTP to an email address.
|
|
"""
|
|
email = kw.get('email')
|
|
phone = kw.get('phone')
|
|
phone_or_email = kw.get('phone_or_email')
|
|
otp_type = kw.get('type')
|
|
|
|
if not otp_type:
|
|
return {'status': 'error', 'message': 'OTP type is required.'}
|
|
|
|
if otp_type == 'signup':
|
|
if not email or not phone:
|
|
return {'status': 'error', 'message': 'Email and Phone are required for sign-up.'}
|
|
# Check if partner or user already exists
|
|
candidates = normalize_phone_search(phone)
|
|
existing_partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
if existing_partner:
|
|
return {'status': 'error', 'message': 'A member with this phone number is already registered.'}
|
|
|
|
existing_user = request.env['res.users'].sudo().with_context(active_test=False).search([
|
|
'|', ('login', '=', phone), ('login', 'in', candidates), ('email', '=', email)
|
|
], limit=1)
|
|
if existing_user:
|
|
return {'status': 'error', 'message': 'A user with this phone number or email already exists.'}
|
|
|
|
elif otp_type == 'activation':
|
|
if not email or not phone:
|
|
return {'status': 'error', 'message': 'Email and Phone are required for activation.'}
|
|
# Check if partner exists
|
|
candidates = normalize_phone_search(phone)
|
|
partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
if not partner:
|
|
return {'status': 'error', 'message': 'Member with this phone number not found. Please contact support or register.'}
|
|
|
|
# Check if user already active
|
|
user = request.env['res.users'].sudo().with_context(active_test=False).search([('partner_id', '=', partner.id)], limit=1)
|
|
if user and user.active:
|
|
return {'status': 'error', 'message': 'Account is already active. Please login or reset password.'}
|
|
|
|
# Check if email is already taken by another active user
|
|
existing_email_user = request.env['res.users'].sudo().search([('email', '=', email), ('partner_id', '!=', partner.id)], limit=1)
|
|
if existing_email_user:
|
|
return {'status': 'error', 'message': 'This email address is already registered to another account.'}
|
|
|
|
elif otp_type == 'reset_password':
|
|
if not phone_or_email:
|
|
return {'status': 'error', 'message': 'Phone number or Email is required.'}
|
|
|
|
# Find user
|
|
user = None
|
|
if '@' in phone_or_email:
|
|
user = request.env['res.users'].sudo().search([('email', '=', phone_or_email)], limit=1)
|
|
else:
|
|
candidates = normalize_phone_search(phone_or_email)
|
|
user = request.env['res.users'].sudo().search([('login', 'in', candidates)], limit=1)
|
|
if not user:
|
|
partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
if partner:
|
|
user = request.env['res.users'].sudo().search([('partner_id', '=', partner.id)], limit=1)
|
|
|
|
if not user:
|
|
return {'status': 'error', 'message': 'User account not found.'}
|
|
|
|
email = user.email
|
|
if not email or '@miemapan.com' in email or '@' not in email:
|
|
return {'status': 'error', 'message': 'Your account does not have a valid email set. Please contact support.'}
|
|
|
|
phone = user.login
|
|
|
|
try:
|
|
request.env['loyalty.verification.otp'].sudo().generate_otp(email=email, phone=phone, otp_type=otp_type)
|
|
# Mask email for privacy response
|
|
parts = email.split('@')
|
|
masked_email = parts[0][:2] + '*' * (len(parts[0]) - 2) + '@' + parts[1] if len(parts[0]) > 2 else email
|
|
return {
|
|
'status': 'success',
|
|
'message': f'Verification code sent to {masked_email}.',
|
|
'email': email
|
|
}
|
|
except Exception as e:
|
|
return {'status': 'error', 'message': f'Failed to send verification code: {str(e)}'}
|
|
|
|
@http.route('/api/loyalty/activate_account', type='jsonrpc', auth='public', methods=['POST'], csrf=False)
|
|
def activate_account(self, **kw):
|
|
"""
|
|
Public endpoint to activate/set password for an existing partner.
|
|
"""
|
|
phone = kw.get('phone')
|
|
email = kw.get('email')
|
|
password = kw.get('password')
|
|
birth_date = kw.get('birth_date')
|
|
otp = kw.get('otp')
|
|
|
|
if not phone or not password or not email or not otp:
|
|
return {'status': 'error', 'message': 'Phone, Email, Password, and Verification Code (OTP) are required.'}
|
|
|
|
# Verify OTP first
|
|
otp_verified = request.env['loyalty.verification.otp'].sudo().verify_otp(
|
|
otp_code=otp, email=email, phone=phone, otp_type='activation'
|
|
)
|
|
if not otp_verified:
|
|
return {'status': 'error', 'message': 'Invalid or expired verification code.'}
|
|
|
|
candidates = normalize_phone_search(phone)
|
|
partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
|
|
if not partner:
|
|
return {'status': 'error', 'message': 'Member with this phone number not found. Please contact support or register.'}
|
|
|
|
# Verify birth date if set on the partner
|
|
if partner.birth_date:
|
|
if not birth_date:
|
|
return {'status': 'error', 'message': 'Birth Date is required for verification.'}
|
|
if str(partner.birth_date) != str(birth_date):
|
|
return {'status': 'error', 'message': 'Birth Date does not match our records.'}
|
|
|
|
# Update email on partner
|
|
partner.sudo().write({'email': email})
|
|
|
|
# Find or create user
|
|
user = request.env['res.users'].sudo().with_context(active_test=False).search([('partner_id', '=', partner.id)], limit=1)
|
|
if user:
|
|
user.sudo().write({
|
|
'company_id': request.env.company.id,
|
|
'company_ids': [(6, 0, [request.env.company.id])],
|
|
'login': phone,
|
|
'email': email,
|
|
'password': password,
|
|
'active': True
|
|
})
|
|
portal_group = request.env.ref('base.group_portal')
|
|
if portal_group not in user.group_ids:
|
|
user.sudo().write({'group_ids': [(4, portal_group.id)]})
|
|
else:
|
|
portal_template = request.env.ref('base.template_portal_user_id')
|
|
existing_user = request.env['res.users'].sudo().with_context(active_test=False).search([('login', '=', phone)], limit=1)
|
|
if existing_user:
|
|
return {'status': 'error', 'message': 'A user with this phone number login already exists.'}
|
|
|
|
try:
|
|
user = portal_template.sudo().with_context(no_reset_password=True).copy({
|
|
'name': partner.name,
|
|
'login': phone,
|
|
'email': email,
|
|
'partner_id': partner.id,
|
|
'active': False,
|
|
})
|
|
# Write company, password and activate
|
|
user.sudo().write({
|
|
'company_id': request.env.company.id,
|
|
'company_ids': [(6, 0, [request.env.company.id])],
|
|
'password': password,
|
|
'active': True,
|
|
})
|
|
# Ensure portal group
|
|
portal_group = request.env.ref('base.group_portal')
|
|
if portal_group not in user.group_ids:
|
|
user.sudo().write({'group_ids': [(4, portal_group.id)]})
|
|
except Exception as e:
|
|
return {'status': 'error', 'message': f'Failed to create user: {str(e)}'}
|
|
|
|
return {'status': 'success', 'message': 'Account activated successfully. You can now login.'}
|
|
|
|
@http.route('/api/loyalty/signup_member', type='jsonrpc', auth='public', methods=['POST'], csrf=False)
|
|
def signup_member(self, **kw):
|
|
"""
|
|
Public endpoint to sign up a new member.
|
|
"""
|
|
name = kw.get('name')
|
|
phone = kw.get('phone')
|
|
email = kw.get('email')
|
|
birth_date = kw.get('birth_date')
|
|
gender = kw.get('gender')
|
|
password = kw.get('password')
|
|
otp = kw.get('otp')
|
|
|
|
if not name or not phone or not email or not password or not otp:
|
|
return {'status': 'error', 'message': 'Name, Phone, Email, Password, and Verification Code (OTP) are required.'}
|
|
|
|
# Verify OTP first
|
|
otp_verified = request.env['loyalty.verification.otp'].sudo().verify_otp(
|
|
otp_code=otp, email=email, phone=phone, otp_type='signup'
|
|
)
|
|
if not otp_verified:
|
|
return {'status': 'error', 'message': 'Invalid or expired verification code.'}
|
|
|
|
candidates = normalize_phone_search(phone)
|
|
|
|
existing_partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
if existing_partner:
|
|
return {'status': 'error', 'message': 'A member with this phone number is already registered.'}
|
|
|
|
existing_user = request.env['res.users'].sudo().with_context(active_test=False).search([
|
|
'|', ('login', '=', phone), ('login', 'in', candidates), ('email', '=', email)
|
|
], limit=1)
|
|
if existing_user:
|
|
return {'status': 'error', 'message': 'A user with this phone number or email already exists.'}
|
|
|
|
try:
|
|
partner_vals = {
|
|
'name': name,
|
|
'phone': phone,
|
|
'email': email,
|
|
'birth_date': birth_date,
|
|
'gender': gender,
|
|
'company_id': request.env.company.id,
|
|
}
|
|
partner = request.env['res.partner'].sudo().create(partner_vals)
|
|
|
|
portal_template = request.env.ref('base.template_portal_user_id')
|
|
user = portal_template.sudo().with_context(no_reset_password=True).copy({
|
|
'name': partner.name,
|
|
'login': phone,
|
|
'email': email,
|
|
'partner_id': partner.id,
|
|
'active': False,
|
|
})
|
|
# Write company, password and activate
|
|
user.sudo().write({
|
|
'company_id': request.env.company.id,
|
|
'company_ids': [(6, 0, [request.env.company.id])],
|
|
'password': password,
|
|
'active': True,
|
|
})
|
|
# Ensure portal group
|
|
portal_group = request.env.ref('base.group_portal')
|
|
if portal_group not in user.group_ids:
|
|
user.sudo().write({'group_ids': [(4, portal_group.id)]})
|
|
|
|
return {'status': 'success', 'message': 'Account registered successfully.'}
|
|
except Exception as e:
|
|
return {'status': 'error', 'message': f'Registration failed: {str(e)}'}
|
|
|
|
@http.route('/api/loyalty/reset_password', type='jsonrpc', auth='public', methods=['POST'], csrf=False)
|
|
def reset_password(self, **kw):
|
|
"""
|
|
Public endpoint to reset user password using verification OTP.
|
|
"""
|
|
phone_or_email = kw.get('phone_or_email')
|
|
otp = kw.get('otp')
|
|
new_password = kw.get('password')
|
|
|
|
if not phone_or_email or not otp or not new_password:
|
|
return {'status': 'error', 'message': 'Phone/Email, OTP, and New Password are required.'}
|
|
|
|
# Find user
|
|
user = None
|
|
if '@' in phone_or_email:
|
|
user = request.env['res.users'].sudo().search([('email', '=', phone_or_email)], limit=1)
|
|
else:
|
|
candidates = normalize_phone_search(phone_or_email)
|
|
user = request.env['res.users'].sudo().search([('login', 'in', candidates)], limit=1)
|
|
if not user:
|
|
partner = request.env['res.partner'].sudo().search([
|
|
'|', ('phone', 'in', candidates), ('phone_sanitized', 'in', candidates)
|
|
], limit=1)
|
|
if partner:
|
|
user = request.env['res.users'].sudo().search([('partner_id', '=', partner.id)], limit=1)
|
|
|
|
if not user:
|
|
return {'status': 'error', 'message': 'User account not found.'}
|
|
|
|
# Verify OTP
|
|
otp_verified = request.env['loyalty.verification.otp'].sudo().verify_otp(
|
|
otp_code=otp, email=user.email, phone=user.login, otp_type='reset_password'
|
|
)
|
|
if not otp_verified:
|
|
return {'status': 'error', 'message': 'Invalid or expired verification code.'}
|
|
|
|
try:
|
|
user.sudo().write({
|
|
'password': new_password,
|
|
'active': True
|
|
})
|
|
return {'status': 'success', 'message': 'Password has been reset successfully.'}
|
|
except Exception as e:
|
|
return {'status': 'error', 'message': f'Failed to reset password: {str(e)}'}
|
|
|
|
@http.route('/api/loyalty/delete_account', type='jsonrpc', auth='user', methods=['POST'], csrf=False)
|
|
def delete_account(self, **kw):
|
|
"""
|
|
Authenticated endpoint to permanently delete/anonymize a member account.
|
|
"""
|
|
user = request.env.user
|
|
password = kw.get('password')
|
|
|
|
if not password:
|
|
return {'status': 'error', 'message': 'Password is required.'}
|
|
|
|
try:
|
|
user.with_user(user)._check_credentials({'type': 'password', 'password': password}, {'interactive': True})
|
|
except Exception:
|
|
return {'status': 'error', 'message': 'Incorrect password.'}
|
|
|
|
partner = user.partner_id
|
|
|
|
user.sudo().write({'active': False})
|
|
|
|
partner.sudo().write({
|
|
'name': f'Deleted Member {partner.id}',
|
|
'phone': False,
|
|
'email': False,
|
|
'birth_date': False,
|
|
'gender': False,
|
|
'active': False,
|
|
})
|
|
|
|
partner.loyalty_card_ids.sudo().write({
|
|
'active': False,
|
|
'points': 0.0
|
|
})
|
|
|
|
return {'status': 'success', 'message': 'Account deleted successfully.'}
|