from odoo import models, api, fields
import time
from datetime import date,datetime
from odoo.exceptions import ValidationError
from math import isnan
import io
import xlsxwriter
from odoo.tools.misc import xlsxwriter
import base64
class Sequence_Generator(models.AbstractModel):
_name = 'sos_common_scripts'
_description = 'Utility for Generating Custom Sequences'
@api.model
def generate_sequence(self, model_name, form_name, field_name):
today = fields.Date.today()
year_start = today.year if today.month > 3 else today.year - 1
year_end = year_start + 1
fy = f"{year_start % 100}-{year_end % 100}"
month = today.strftime('%m')
base_sequence_prefix = f"SOS/{form_name}/{fy}/{month}/"
records = self.env[model_name].sudo().search(
[(field_name, 'like', f"{base_sequence_prefix}%")],
order=f"{field_name} desc",
limit=1
)
if records and records[0][field_name]:
last_sequence = records[0][field_name]
try:
last_number = int(last_sequence.split('/')[-1])
new_number = last_number + 1
except ValueError:
new_number = 1
else:
new_number = 1
return f"{base_sequence_prefix}{new_number:03d}"
@staticmethod
def _write_sheet(workbook, sheet_name, line_records, headers):
"""Write data to each sheet based on specified fields and custom headers."""
worksheet = workbook.add_worksheet(sheet_name)
# Define Excel date format
date_format = workbook.add_format({'num_format': 'dd/mm/yyyy'})
# Write headers
for col, header in enumerate(headers.values()):
worksheet.write(0, col, header)
# Write data
for row, record in enumerate(line_records, start=1):
for col, field in enumerate(headers.keys()):
value = getattr(record, field, '')
if isinstance(value, models.Model):
worksheet.write(row, col, value.display_name if value else '')
elif field in ['date', 'next_action_date'] and value:
if isinstance(value, (fields.Date, fields.Datetime)):
worksheet.write(row, col, value, date_format)
else:
worksheet.write(row, col, str(value) if value not in (False, None, '') else '')
else:
worksheet.write(row, col, value if value not in (False, None, '') else '')
# def _write_sheet(self, workbook, sheet_name, line_records, headers):
# """Write data to an Excel sheet."""
# worksheet = workbook.add_worksheet(sheet_name)
# # Write headers to the first row
# for col, header in enumerate(headers.values()):
# worksheet.write(0, col, header)
# # Write record data starting from the second row
# for row, record in enumerate(line_records, start=1):
# for col, field in enumerate(headers.keys()):
# value = getattr(record, field, '')
# if isinstance(value, models.Model):
# worksheet.write(row, col, value.display_name if value else '')
# elif isinstance(value, (fields.Date, fields.Datetime)):
# worksheet.write(row, col, value.strftime('%Y-%m-%d') if value else '')
# else:
# worksheet.write(row, col, value)
# def action_export_common(self, headers, column, heading=None):
# """Generate an Excel file with the provided headers and data."""
# output = io.BytesIO()
# workbook = xlsxwriter.Workbook(output, {'in_memory': True})
# # Write data to the workbook
# self._write_sheet(workbook, heading, column, headers)
# workbook.close()
# output.seek(0)
# print(column)
# for record in column:
# attachment = self.env['ir.attachment'].create({
# 'name': 'Material_Procurement_Plan_Export.xlsx',
# 'type': 'binary',
# 'datas': base64.b64encode(output.read()),
# 'res_model': record._name,
# 'res_id': record.id if hasattr(record, 'id') else None,
# 'mimetype': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
# })
# return {
# 'type': 'ir.actions.act_url',
# 'url': f'/web/content/{attachment.id}?download=true',
# 'target': 'self',
# }
@staticmethod
def action_assign_signature(record, user_field_name, approved_on_field_name=None, approved_dept=None, no_signature_message=None):
current_user = record.env.user
esign = current_user.signature_image
# Bypass checks for users in the `sos_inventory.sos_management_user` group
if current_user.has_group('sos_inventory.sos_management_user'):
# Directly assign the signature and exit the method
setattr(record, user_field_name, current_user)
if approved_on_field_name and hasattr(record, approved_on_field_name):
setattr(record, approved_on_field_name, datetime.now())
return None
if approved_dept:
if isinstance(approved_dept, int):
component = record.env['sos_mon'].browse(approved_dept)
users = record.env['res.users'].search([('id', '=', component.filled_by.id)])
reporting_user = record.env['res.users'].search([('id', '=', users.reporting_to.id)])
if current_user.id == reporting_user.id:
pass
else:
raise ValidationError("You do not have the necessary authorization to sign here.")
else:
if not current_user.has_group(approved_dept):
raise ValidationError("You do not have the necessary authorization to sign here.")
if not esign:
message = no_signature_message or "Your signature is missing. Please upload your signature to proceed."
raise ValidationError(message)
setattr(record, user_field_name, current_user)
if approved_on_field_name:
if hasattr(record, approved_on_field_name):
setattr(record, approved_on_field_name, datetime.now())
else:
record.env.cr.rollback()
raise AttributeError(f"Field '{approved_on_field_name}' does not exist on the record.")
return None
@staticmethod
def calculate_week_number(start_date, end_date, input_date):
if not start_date or not end_date or not input_date:
raise ValidationError("Without Indent plan you can't do this Opreation")
start_date_dt = datetime.strptime(str(start_date), '%Y-%m-%d')
end_date_dt = datetime.strptime(str(end_date), '%Y-%m-%d')
input_date_dt = datetime.strptime(str(input_date), '%Y-%m-%d')
if input_date_dt < start_date_dt or input_date_dt > end_date_dt:
return 8
# raise ValidationError("Date exceeds the target date range.")
else:
days_diff = (input_date_dt - start_date_dt).days
week_number = days_diff // 7 + 1
return week_number
@staticmethod
def send_group_email(env,model, record_id, email_to, subject, message,group_xml_id):
group = env.ref(group_xml_id)
action = env['ir.actions.act_window'].search([('res_model', '=', model)], limit=1)
action_id = action.id
menu = env['ir.ui.menu'].search([('action', '=', f'ir.actions.act_window,{action_id}')], limit=1)
menu_id = menu.id
url = f"https://slink.sosaley.in//web#id={record_id}&cids=1&menu_id={menu_id}&action={action_id}&model={model}&view_type=form"
body_html = f"""
Hello User,
{message}
Click the button below to proceed:
"""
if group:
users = env['res.users'].search([('groups_id', 'in', [group.id])])
emails = users.mapped('login')
email_to = ','.join(emails)
mail_values = {
'subject': subject,
'body_html': body_html,
'email_to': email_to,
'email_from': "slink ",
}
mail = env['mail.mail'].sudo().create(mail_values)
mail.sudo().send()
else:
print("No Recepients Found")
@staticmethod
def send_direct_email(env,model, record_id, email_to, subject, message, cc_mail=None):
action = env['ir.actions.act_window'].search([('res_model', '=', model)], limit=1)
action_id = action.id
# Find the menu_id associated with this action
menu = env['ir.ui.menu'].search([('action', '=', f'ir.actions.act_window,{action_id}')], limit=1)
menu_id = menu.id
url = f"https://slink.sosaley.in//web#id={record_id}&cids=1&menu_id={menu_id}&action={action_id}&model={model}&view_type=form"
body_html = f"""
Hello User,
{message}
Click the button below to proceed:
"""
if email_to:
mail_values = {
'subject': subject,
'body_html': body_html,
'email_to': email_to,
'email_cc': cc_mail,
'email_from': "slink ",
}
mail = env['mail.mail'].sudo().create(mail_values)
mail.sudo().send()
else:
print("No Recepients Found")
@staticmethod
def send_mon_min_email(env,model, record_id, email_to, subject, body,user_type):
action = env['ir.actions.act_window'].search([('res_model', '=', model)], limit=1)
action_id = action.id
# Find the menu_id associated with this action
menu = env['ir.ui.menu'].search([('action', '=', f'ir.actions.act_window,{action_id}')], limit=1)
menu_id = menu.id
url = f"https://slink.sosaley.in//web#id={record_id}&cids=1&menu_id={menu_id}&action={action_id}&model={model}&view_type=form"
body_html = f"""
Hello User,
{body}
Click the button below to proceed:
"""
if user_type == "user":
users = env['res.users'].search([('id', '=', env.user.id)])
reporting_user = env['res.users'].search([('id', '=', users.reporting_to.id)])
email_to = reporting_user.login
else:
email_to = "ramachandran.r@sosaley.in"
mail_values = {
'subject': subject,
'body_html': body_html,
'email_to': email_to,
'email_from': "slink ",
}
mail = env['mail.mail'].sudo().create(mail_values)
mail.sudo().send()
@staticmethod
def update_field_value(self, model_name, record_id, field_name, value_to_add):
record = self.env[model_name].search([('id', '=', record_id)], limit=1)
if record:
current_value = getattr(record, field_name, 0)
record.write({field_name: current_value + value_to_add})
return True
return False
@staticmethod
def extract_value(input_string):
total = 0
items = input_string.split(",") # handles both single and comma-separated values
for item in items:
parts = item.split("-")
if len(parts) > 1:
try:
total += int(parts[1])
except ValueError:
total += 0 # skip non-numeric values
else:
total += 0 # no hyphen or second part
return total