# -*- coding: utf-8 -*-
import base64
import os
import io
from datetime import date
from odoo import models, fields, api, tools
from odoo.exceptions import UserError
class HrT4Summary(models.Model):
"""T4 Summary - One per company per tax year"""
_name = 'hr.t4.summary'
_description = 'T4 Summary'
_order = 'tax_year desc'
_inherit = ['mail.thread', 'mail.activity.mixin']
def _get_pdf_text_coordinates(self):
"""Get text overlay coordinates for flattened PDF
Returns dict mapping field names to (x, y, font_size, font_name) tuples
Coordinates are in points (1/72 inch), origin at bottom-left
Reads from pdf.field.position model based on template type
"""
# Query configured positions from database for T4 Summary
position_model = self.env['pdf.field.position']
return position_model.get_coordinates_dict('T4 Summary')
STATE_SELECTION = [
('draft', 'Draft'),
('generated', 'Generated'),
('filed', 'Filed'),
]
name = fields.Char(
string='Reference',
compute='_compute_name',
store=True,
)
company_id = fields.Many2one(
'res.company',
string='Company',
required=True,
default=lambda self: self.env.company,
)
currency_id = fields.Many2one(
related='company_id.currency_id',
)
tax_year = fields.Integer(
string='Tax Year',
required=True,
default=lambda self: date.today().year - 1,
)
state = fields.Selection(
selection=STATE_SELECTION,
string='Status',
default='draft',
tracking=True,
)
# === CRA Information ===
cra_business_number = fields.Char(
string='CRA Business Number',
compute='_compute_cra_business_number',
readonly=True,
)
@api.depends('company_id')
def _compute_cra_business_number(self):
"""Get CRA business number from payroll settings."""
for summary in self:
if summary.company_id:
settings = self.env['payroll.config.settings'].get_settings(summary.company_id.id)
summary.cra_business_number = settings.get_cra_payroll_account_number() or summary.company_id.vat or ''
else:
summary.cra_business_number = ''
# === Box 88: Number of T4 Slips ===
slip_count = fields.Integer(
string='Total T4 Slips (Box 88)',
compute='_compute_totals',
store=True,
)
slip_ids = fields.One2many(
'hr.t4.slip',
'summary_id',
string='T4 Slips',
)
# === Box 14: Employment Income ===
total_employment_income = fields.Monetary(
string='Employment Income (Box 14)',
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 16: Employees CPP Contributions ===
total_cpp_employee = fields.Monetary(
string="Employees' CPP (Box 16)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 16A: Employees CPP2 Contributions ===
total_cpp2_employee = fields.Monetary(
string="Employees' CPP2 (Box 16A)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 18: Employees EI Premiums ===
total_ei_employee = fields.Monetary(
string="Employees' EI (Box 18)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 19: Employer EI Premiums ===
total_ei_employer = fields.Monetary(
string="Employer's EI (Box 19)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 22: Income Tax Deducted ===
total_income_tax = fields.Monetary(
string='Income Tax (Box 22)',
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 27: Employer CPP Contributions ===
total_cpp_employer = fields.Monetary(
string="Employer's CPP (Box 27)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 27A: Employer CPP2 Contributions ===
total_cpp2_employer = fields.Monetary(
string="Employer's CPP2 (Box 27A)",
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 80: Total Deductions ===
total_deductions = fields.Monetary(
string='Total Deductions (Box 80)',
currency_field='currency_id',
compute='_compute_totals',
store=True,
)
# === Box 82: Remittances ===
total_remittances = fields.Monetary(
string='Total Remittances (Box 82)',
currency_field='currency_id',
)
# === Box 84/86: Overpayment/Balance Due ===
difference = fields.Monetary(
string='Difference',
currency_field='currency_id',
compute='_compute_difference',
store=True,
)
overpayment = fields.Monetary(
string='Overpayment (Box 84)',
currency_field='currency_id',
compute='_compute_difference',
store=True,
)
balance_due = fields.Monetary(
string='Balance Due (Box 86)',
currency_field='currency_id',
compute='_compute_difference',
store=True,
)
# === Contact Information ===
contact_name = fields.Char(
string='Contact Person (Box 76)',
default=lambda self: self.env.user.name,
)
contact_phone = fields.Char(
string='Telephone (Box 78)',
)
# === Filing Information ===
filing_date = fields.Date(
string='Filing Date',
tracking=True,
)
xml_file = fields.Binary(
string='XML File',
attachment=True,
)
xml_filename = fields.Char(
string='XML Filename',
)
# === Box 74: SIN of Proprietor ===
proprietor_sin = fields.Char(
string='SIN of Proprietor (Box 74)',
help='Social Insurance Number of proprietor(s) or principal owner(s)',
)
# === PDF Generation ===
filled_pdf = fields.Binary(
string='Filled PDF',
attachment=True,
)
filled_pdf_filename = fields.Char(
string='PDF Filename',
)
@api.depends('tax_year', 'company_id')
def _compute_name(self):
for rec in self:
rec.name = f"T4 Summary {rec.tax_year} - {rec.company_id.name}"
@api.depends('slip_ids', 'slip_ids.employment_income', 'slip_ids.cpp_employee',
'slip_ids.cpp2_employee', 'slip_ids.ei_employee', 'slip_ids.income_tax')
def _compute_totals(self):
for rec in self:
slips = rec.slip_ids
rec.slip_count = len(slips)
rec.total_employment_income = sum(slips.mapped('employment_income'))
rec.total_cpp_employee = sum(slips.mapped('cpp_employee'))
rec.total_cpp2_employee = sum(slips.mapped('cpp2_employee'))
rec.total_ei_employee = sum(slips.mapped('ei_employee'))
rec.total_income_tax = sum(slips.mapped('income_tax'))
rec.total_cpp_employer = sum(slips.mapped('cpp_employer'))
rec.total_cpp2_employer = sum(slips.mapped('cpp2_employer'))
rec.total_ei_employer = sum(slips.mapped('ei_employer'))
# Box 80 = 16 + 16A + 27 + 27A + 18 + 19 + 22
rec.total_deductions = (
rec.total_cpp_employee + rec.total_cpp2_employee +
rec.total_cpp_employer + rec.total_cpp2_employer +
rec.total_ei_employee + rec.total_ei_employer +
rec.total_income_tax
)
@api.depends('total_deductions', 'total_remittances')
def _compute_difference(self):
for rec in self:
rec.difference = rec.total_deductions - rec.total_remittances
if rec.difference > 0:
rec.balance_due = rec.difference
rec.overpayment = 0
else:
rec.overpayment = abs(rec.difference)
rec.balance_due = 0
def action_generate_slips(self):
"""Generate T4 slips for all employees with payslips in the tax year"""
self.ensure_one()
# Find all employees with confirmed payslips in the year
year_start = date(self.tax_year, 1, 1)
year_end = date(self.tax_year, 12, 31)
payslips = self.env['hr.payslip'].search([
('company_id', '=', self.company_id.id),
('state', 'in', ['validated', 'paid']),
('date_from', '>=', year_start),
('date_to', '<=', year_end),
])
if not payslips:
raise UserError(f'No confirmed payslips found for {self.tax_year}.')
# Group by employee
employee_payslips = {}
for ps in payslips:
if ps.employee_id.id not in employee_payslips:
employee_payslips[ps.employee_id.id] = []
employee_payslips[ps.employee_id.id].append(ps)
# Delete existing slips
self.slip_ids.unlink()
# Create slip for each employee
T4Slip = self.env['hr.t4.slip']
Payslip = self.env['hr.payslip']
for employee_id, emp_payslips in employee_payslips.items():
employee = self.env['hr.employee'].browse(employee_id)
# Sum amounts from payslips
employment_income = 0
cpp_ee = cpp_er = cpp2_ee = cpp2_er = 0
ei_ee = ei_er = 0
income_tax = 0
# New boxes
box_40_allowances = 0
box_42_commissions = 0
box_44_union_dues = 0
for ps in emp_payslips:
# Process each payslip line
gross_amount = 0
for line in ps.line_ids:
code = line.code or ''
category_code = line.category_id.code if line.category_id else None
amount = abs(line.total or 0)
# Use pay type helpers to identify pay types
pay_type = Payslip._get_pay_type_from_code(code, category_code)
is_reimbursement = Payslip._is_reimbursement(code, category_code)
# Skip reimbursements - they are non-taxable and don't appear on T4
if is_reimbursement:
continue
# Get GROSS amount (this is the total taxable income for Box 14)
if code == 'GROSS' and category_code == 'GROSS':
gross_amount = amount
# Track allowances for Box 40 (these are included in GROSS)
# Only count if it's in income categories, not if it's already in GROSS
if pay_type == 'allowance' and category_code in ['BASIC', 'ALW']:
# Track separately for Box 40, but don't add to employment_income
# because GROSS already includes it
box_40_allowances += amount
# Track commissions for Box 42 (these are included in GROSS)
elif pay_type == 'commission' and category_code in ['BASIC', 'ALW']:
# Track separately for Box 42, but don't add to employment_income
# because GROSS already includes it
box_42_commissions += amount
# Track union dues for Box 44 (deductions, NOT in GROSS)
elif pay_type == 'union_dues' and category_code == 'DED':
box_44_union_dues += amount
# NOT included in Box 14 (it's a deduction, not income)
# Tax deductions
elif code == 'CPP_EE':
cpp_ee += amount
elif code == 'CPP_ER':
cpp_er += amount
elif code == 'CPP2_EE':
cpp2_ee += amount
elif code == 'CPP2_ER':
cpp2_er += amount
elif code == 'EI_EE':
ei_ee += amount
elif code == 'EI_ER':
ei_er += amount
elif code in ('FED_TAX', 'PROV_TAX'):
income_tax += amount
# Add GROSS to employment income (Box 14)
# GROSS already includes all taxable income (salary, overtime, bonus, allowances, commissions, etc.)
employment_income += gross_amount
# Calculate Box 24 and Box 26
# In most cases, they equal Box 14, but can differ for exempt amounts
# For now, set them equal to Box 14 (can be enhanced later for exempt amounts)
ei_insurable_earnings = employment_income
cpp_pensionable_earnings = employment_income
T4Slip.create({
'summary_id': self.id,
'employee_id': employee_id,
'employment_income': employment_income,
'cpp_employee': cpp_ee,
'cpp_employer': cpp_er,
'cpp2_employee': cpp2_ee,
'cpp2_employer': cpp2_er,
'ei_employee': ei_ee,
'ei_employer': ei_er,
'income_tax': income_tax,
# Box 24 and 26 - must never be blank per CRA
'ei_insurable_earnings': ei_insurable_earnings,
'cpp_pensionable_earnings': cpp_pensionable_earnings,
# New boxes
'box_40_taxable_benefits': box_40_allowances,
'box_42_commissions': box_42_commissions,
'box_44_union_dues': box_44_union_dues,
})
self.state = 'generated'
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'T4 Slips Generated',
'message': f'Generated {len(employee_payslips)} T4 slips.',
'type': 'success',
}
}
def action_mark_filed(self):
"""Mark T4 Summary as filed"""
self.ensure_one()
self.write({
'state': 'filed',
'filing_date': date.today(),
})
def _get_pdf_text_coordinates(self):
"""Get text overlay coordinates for flattened PDF
Returns dict mapping field names to (x, y, font_size, font_name) tuples
Coordinates are in points (1/72 inch), origin at bottom-left
Reads from pdf.field.position model based on template type
"""
# Determine template type from model name
template_type = None
if self._name == 'hr.t4.slip':
template_type = 'T4'
elif self._name == 'hr.t4.summary':
template_type = 'T4 Summary'
if template_type:
# Query configured positions from database
position_model = self.env['pdf.field.position']
return position_model.get_coordinates_dict(template_type)
# Fallback: return empty dict if template type not recognized
return {}
def _overlay_text_on_pdf(self, template_path, field_mapping):
"""Overlay text on a flattened PDF using reportlab
Returns base64-encoded PDF data
"""
try:
from reportlab.pdfgen import canvas
from PyPDF2 import PdfReader, PdfWriter
except ImportError as e:
raise UserError(f'Required library not available: {str(e)}\nPlease install reportlab and PyPDF2.')
# Get text coordinates
text_coords = self._get_pdf_text_coordinates()
if not text_coords:
# Determine template type for error message
template_type = 'T4'
if self._name == 'hr.t4.summary':
template_type = 'T4 Summary'
raise UserError(
f'Text coordinates not configured for {template_type} template. '
'Please configure PDF field positions in Payroll → Configuration → PDF Field Positions.'
)
# Read the template PDF
with open(template_path, 'rb') as template_file:
template_reader = PdfReader(template_file)
if not template_reader.pages:
raise UserError('Template PDF has no pages')
# Get first page dimensions
first_page = template_reader.pages[0]
page_width = float(first_page.mediabox.width)
page_height = float(first_page.mediabox.height)
# Create overlay PDF with text
overlay_buffer = io.BytesIO()
can = canvas.Canvas(overlay_buffer, pagesize=(page_width, page_height))
# Draw text for each field
for field_name, value in field_mapping.items():
if field_name in text_coords and value:
coord_data = text_coords[field_name]
# Handle both old format (x, y, font_size) and new format (x, y, font_size, font_name)
if len(coord_data) == 4:
x, y, font_size, font_name = coord_data
elif len(coord_data) == 3:
x, y, font_size = coord_data
font_name = 'Helvetica' # Default font
else:
continue # Skip invalid coordinate data
can.setFont(font_name, font_size)
can.drawString(x, y, str(value))
can.save()
overlay_buffer.seek(0)
# Merge overlay with template
with open(template_path, 'rb') as template_file:
template_reader = PdfReader(template_file)
overlay_reader = PdfReader(overlay_buffer)
writer = PdfWriter()
for i, page in enumerate(template_reader.pages):
if i < len(overlay_reader.pages):
page.merge_page(overlay_reader.pages[i])
writer.add_page(page)
# Write to bytes
output_buffer = io.BytesIO()
writer.write(output_buffer)
return base64.b64encode(output_buffer.getvalue())
def action_fill_pdf(self):
"""Fill the T4 Summary PDF form with data from this summary"""
self.ensure_one()
try:
# Try to import pdfrw (preferred) or PyPDF2
try:
from pdfrw import PdfReader, PdfWriter, PdfDict
use_pdfrw = True
except ImportError:
try:
import PyPDF2
use_pdfrw = False
except ImportError:
raise UserError(
'PDF library not found. Please install pdfrw or PyPDF2:\n'
'pip install pdfrw\n'
'or\n'
'pip install PyPDF2'
)
# Get PDF template path - try multiple locations
module_path = os.path.dirname(os.path.dirname(__file__))
template_path = os.path.join(module_path, 'static', 'pdf', 't4sum-fill-25e.pdf')
# Try in module root directory (fallback)
if not os.path.exists(template_path):
template_path = os.path.join(module_path, 't4sum-fill-25e.pdf')
# Try using tools.file_path (Odoo 19)
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/static/pdf/t4sum-fill-25e.pdf')
except:
pass
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/t4sum-fill-25e.pdf')
except:
pass
if not os.path.exists(template_path):
raise UserError(
'T4 Summary PDF template not found. Please ensure t4sum-fill-25e.pdf is in one of these locations:\n'
f'1. {os.path.join(module_path, "static", "pdf", "t4sum-fill-25e.pdf")} (recommended)\n'
f'2. {os.path.join(module_path, "t4sum-fill-25e.pdf")} (module root)\n\n'
'The system will automatically fill the PDF with data from this T4 Summary when you click "Fill PDF".'
)
# Get field mapping
field_mapping = self._get_pdf_field_mapping()
# Check if we should use text overlay (for flattened PDFs)
text_coords = self._get_pdf_text_coordinates()
if text_coords:
# Use text overlay method for flattened PDF
pdf_data = self._overlay_text_on_pdf(template_path, field_mapping)
elif use_pdfrw:
# Use pdfrw to fill PDF
try:
template = PdfReader(template_path)
# Fill form fields
if hasattr(template.Root, 'AcroForm') and template.Root.AcroForm:
if hasattr(template.Root.AcroForm, 'Fields') and template.Root.AcroForm.Fields:
for field in template.Root.AcroForm.Fields:
# Get field name (can be in /T or /TU)
field_name = None
if hasattr(field, 'T'):
field_name = str(field.T).strip('()')
elif hasattr(field, 'TU'):
field_name = str(field.TU).strip('()')
if field_name and field_name in field_mapping:
value = field_mapping[field_name]
if value is not None and value != '':
# Set field value
field.V = str(value)
# Make sure field is not read-only
if hasattr(field, 'Ff'):
field.Ff = 0 # Remove read-only flag
# Write filled PDF to temporary file
import tempfile
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:
tmp_path = tmp_file.name
writer = PdfWriter()
writer.write(template, tmp_path)
# Read filled PDF
with open(tmp_path, 'rb') as f:
pdf_data = base64.b64encode(f.read())
# Clean up temp file
try:
os.remove(tmp_path)
except:
pass
except (ValueError, TypeError, Exception) as e:
# pdfrw failed to parse the PDF, fall back to PyPDF2
import logging
_logger = logging.getLogger(__name__)
_logger.warning(f'pdfrw failed to read PDF ({str(e)}), falling back to PyPDF2')
use_pdfrw = False
# Import PyPDF2 for fallback
try:
import PyPDF2
except ImportError:
raise UserError(
f'pdfrw failed to read the PDF file, and PyPDF2 is not available.\n'
f'Error: {str(e)}\n\n'
f'Please install PyPDF2: pip install PyPDF2'
)
if not use_pdfrw:
# Use PyPDF2 (fallback)
with open(template_path, 'rb') as template_file:
reader = PyPDF2.PdfReader(template_file)
writer = PyPDF2.PdfWriter()
# Copy pages
try:
for page in reader.pages:
writer.add_page(page)
except Exception as page_error:
# Check if PyCryptodome is needed
try:
import Cryptodome
except ImportError:
raise UserError(
f'PDF file appears to be encrypted and requires PyCryptodome for decryption.\n'
f'Error: {str(page_error)}\n\n'
f'Please install PyCryptodome: pip install pycryptodome\n'
f'Or use an unencrypted PDF template.'
)
raise
# Fill form fields (PyPDF2 approach)
if '/AcroForm' in reader.trailer['/Root']:
writer.update_page_form_field_values(writer.pages[0], field_mapping)
# Write to bytes
output_buffer = io.BytesIO()
writer.write(output_buffer)
pdf_data = base64.b64encode(output_buffer.getvalue())
# Generate filename
company_name = (self.company_id.name or 'Company').replace(" ", "_").replace("/", "_")[:20]
filename = f'T4Summary_{self.tax_year}_{company_name}.pdf'
# Save filled PDF
self.write({
'filled_pdf': pdf_data,
'filled_pdf_filename': filename,
})
# Create attachment
attachment = self.env['ir.attachment'].create({
'name': filename,
'type': 'binary',
'datas': pdf_data,
'res_model': self._name,
'res_id': self.id,
'mimetype': 'application/pdf',
})
# Post to chatter
self.message_post(
body=f'T4 Summary PDF generated: {filename}',
attachment_ids=[attachment.id],
)
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'PDF Generated',
'message': f'T4 Summary PDF filled and saved: {filename}',
'type': 'success',
}
}
except Exception as e:
import traceback
error_msg = f'Error filling PDF: {str(e)}\n\nTraceback:\n{traceback.format_exc()}'
raise UserError(error_msg)
def action_extract_pdf_fields(self):
"""Helper method to extract PDF form field names (for debugging)"""
self.ensure_one()
try:
from pdfrw import PdfReader
except ImportError:
raise UserError('pdfrw library not installed. Install with: pip install pdfrw')
# Get PDF template path
module_path = os.path.dirname(os.path.dirname(__file__))
template_path = os.path.join(module_path, 'static', 'pdf', 't4sum-fill-25e.pdf')
if not os.path.exists(template_path):
template_path = os.path.join(module_path, 't4sum-fill-25e.pdf')
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/static/pdf/t4sum-fill-25e.pdf')
except:
template_path = None
if not template_path or not os.path.exists(template_path):
raise UserError('T4 Summary PDF template not found.')
template = PdfReader(template_path)
field_names = []
# Extract field names from all pages
for page_num, page in enumerate(template.pages, 1):
if hasattr(page, 'Annots') and page.Annots:
for annot in page.Annots:
if hasattr(annot, 'Subtype') and str(annot.Subtype) == '/Widget':
if hasattr(annot, 'T'):
field_name = str(annot.T).strip('()')
field_names.append(f'Page {page_num}: {field_name}')
# Return as message
if field_names:
message = 'PDF Form Fields Found:\n\n' + '\n'.join(field_names[:50])
if len(field_names) > 50:
message += f'\n\n... and {len(field_names) - 50} more fields'
else:
message = 'No form fields found in PDF. The PDF may not be a fillable form, or field names are stored differently.'
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'PDF Fields',
'message': message,
'type': 'info',
'sticky': True,
}
}
def action_download_pdf(self):
"""Download the filled PDF"""
self.ensure_one()
if not self.filled_pdf:
raise UserError('No PDF has been generated yet. Please click "Fill PDF" first.')
return {
'type': 'ir.actions.act_url',
'url': f'/web/content/hr.t4.summary/{self.id}/filled_pdf/{self.filled_pdf_filename}?download=true',
'target': 'self',
}
def _get_pdf_field_mapping(self):
"""Map model fields to PDF form field names"""
# This mapping may need to be adjusted based on actual PDF form field names
# You can use action_extract_pdf_fields() to see actual field names in the PDF
mapping = {}
# Year
mapping['Year'] = str(self.tax_year)
mapping['year'] = str(self.tax_year)
mapping['YEAR'] = str(self.tax_year)
mapping['TaxYear'] = str(self.tax_year)
mapping['YearEnding'] = str(self.tax_year)
# Payer/Employer information
company = self.company_id
if company:
mapping['PayerName'] = company.name or ''
mapping['PayerName1'] = company.name or ''
mapping['EmployerName'] = company.name or ''
if company.street:
mapping['PayerAddress1'] = company.street
mapping['EmployerAddress1'] = company.street
if company.street2:
mapping['PayerAddress2'] = company.street2
mapping['EmployerAddress2'] = company.street2
if company.city:
mapping['PayerCity'] = company.city
mapping['EmployerCity'] = company.city
if company.state_id:
mapping['PayerProvince'] = company.state_id.code
mapping['EmployerProvince'] = company.state_id.code
if company.zip:
mapping['PayerPostalCode'] = company.zip
mapping['EmployerPostalCode'] = company.zip
# Employer account number (Box 54)
settings = self.env['payroll.config.settings'].get_settings(company.id)
account_num = settings.get_cra_payroll_account_number() or company.vat or ''
mapping['PayerAccount'] = account_num
mapping['Box54'] = account_num
mapping['EmployerAccount'] = account_num
mapping['AccountNumber'] = account_num
# Box 88: Total number of T4 slips
if self.slip_count:
mapping['Box88'] = str(self.slip_count)
mapping['088'] = str(self.slip_count)
mapping['TotalSlips'] = str(self.slip_count)
# Box 14: Employment Income
if self.total_employment_income:
mapping['Box14'] = f"{self.total_employment_income:.2f}"
mapping['014'] = f"{self.total_employment_income:.2f}"
mapping['EmploymentIncome'] = f"{self.total_employment_income:.2f}"
# Box 16: CPP Employee
if self.total_cpp_employee:
mapping['Box16'] = f"{self.total_cpp_employee:.2f}"
mapping['016'] = f"{self.total_cpp_employee:.2f}"
mapping['CPPEmployee'] = f"{self.total_cpp_employee:.2f}"
# Box 16A: CPP2 Employee
if self.total_cpp2_employee:
mapping['Box16A'] = f"{self.total_cpp2_employee:.2f}"
mapping['016A'] = f"{self.total_cpp2_employee:.2f}"
mapping['CPP2Employee'] = f"{self.total_cpp2_employee:.2f}"
# Box 18: EI Employee
if self.total_ei_employee:
mapping['Box18'] = f"{self.total_ei_employee:.2f}"
mapping['018'] = f"{self.total_ei_employee:.2f}"
mapping['EIEmployee'] = f"{self.total_ei_employee:.2f}"
# Box 19: EI Employer
if self.total_ei_employer:
mapping['Box19'] = f"{self.total_ei_employer:.2f}"
mapping['019'] = f"{self.total_ei_employer:.2f}"
mapping['EIEmployer'] = f"{self.total_ei_employer:.2f}"
# Box 22: Income Tax
if self.total_income_tax:
mapping['Box22'] = f"{self.total_income_tax:.2f}"
mapping['022'] = f"{self.total_income_tax:.2f}"
mapping['IncomeTax'] = f"{self.total_income_tax:.2f}"
# Box 27: CPP Employer
if self.total_cpp_employer:
mapping['Box27'] = f"{self.total_cpp_employer:.2f}"
mapping['027'] = f"{self.total_cpp_employer:.2f}"
mapping['CPPEmployer'] = f"{self.total_cpp_employer:.2f}"
# Box 27A: CPP2 Employer
if self.total_cpp2_employer:
mapping['Box27A'] = f"{self.total_cpp2_employer:.2f}"
mapping['027A'] = f"{self.total_cpp2_employer:.2f}"
mapping['CPP2Employer'] = f"{self.total_cpp2_employer:.2f}"
# Box 80: Total Deductions
if self.total_deductions:
mapping['Box80'] = f"{self.total_deductions:.2f}"
mapping['080'] = f"{self.total_deductions:.2f}"
mapping['TotalDeductions'] = f"{self.total_deductions:.2f}"
# Box 82: Remittances
if self.total_remittances:
mapping['Box82'] = f"{self.total_remittances:.2f}"
mapping['082'] = f"{self.total_remittances:.2f}"
mapping['Remittances'] = f"{self.total_remittances:.2f}"
# Box 84: Overpayment
if self.overpayment:
mapping['Box84'] = f"{self.overpayment:.2f}"
mapping['084'] = f"{self.overpayment:.2f}"
mapping['Overpayment'] = f"{self.overpayment:.2f}"
# Box 86: Balance Due
if self.balance_due:
mapping['Box86'] = f"{self.balance_due:.2f}"
mapping['086'] = f"{self.balance_due:.2f}"
mapping['BalanceDue'] = f"{self.balance_due:.2f}"
# Box 74: SIN of Proprietor
if self.proprietor_sin:
sin_clean = self.proprietor_sin.replace('-', '').replace(' ', '')
mapping['Box74'] = sin_clean
mapping['074'] = sin_clean
mapping['ProprietorSIN'] = sin_clean
# Box 76: Contact Name
if self.contact_name:
mapping['Box76'] = self.contact_name
mapping['076'] = self.contact_name
mapping['ContactName'] = self.contact_name
# Box 78: Contact Phone
if self.contact_phone:
# Parse phone number (format: 905-451-7743 or 9054517743)
phone_clean = self.contact_phone.replace('-', '').replace(' ', '').replace('(', '').replace(')', '')
if len(phone_clean) >= 10:
area_code = phone_clean[:3]
number = phone_clean[3:]
mapping['Box78AreaCode'] = area_code
mapping['Box78Phone'] = number
mapping['ContactAreaCode'] = area_code
mapping['ContactPhone'] = number
mapping['Box78'] = self.contact_phone
mapping['078'] = self.contact_phone
return mapping
class HrT4Slip(models.Model):
"""T4 Slip - One per employee per tax year"""
_name = 'hr.t4.slip'
_description = 'T4 Slip'
_order = 'employee_id'
_inherit = ['mail.thread', 'mail.activity.mixin']
summary_id = fields.Many2one(
'hr.t4.summary',
string='T4 Summary',
required=True,
ondelete='cascade',
)
company_id = fields.Many2one(
related='summary_id.company_id',
)
currency_id = fields.Many2one(
related='summary_id.currency_id',
)
tax_year = fields.Integer(
related='summary_id.tax_year',
)
employee_id = fields.Many2one(
'hr.employee',
string='Employee',
required=True,
)
# === Employee Information ===
sin_number = fields.Char(
string='SIN',
related='employee_id.sin_number',
)
employee_name = fields.Char(
string='Employee Name',
related='employee_id.name',
)
# === Box 14: Employment Income ===
employment_income = fields.Monetary(
string='Employment Income (Box 14)',
currency_field='currency_id',
)
# === Box 16: CPP Contributions ===
cpp_employee = fields.Monetary(
string='CPP (Box 16)',
currency_field='currency_id',
)
# === Box 16A: CPP2 Contributions ===
cpp2_employee = fields.Monetary(
string='CPP2 (Box 16A)',
currency_field='currency_id',
)
# === Box 18: EI Premiums ===
ei_employee = fields.Monetary(
string='EI (Box 18)',
currency_field='currency_id',
)
# === Box 22: Income Tax Deducted ===
income_tax = fields.Monetary(
string='Income Tax (Box 22)',
currency_field='currency_id',
)
# === Box 24: EI Insurable Earnings ===
ei_insurable_earnings = fields.Monetary(
string='EI Insurable Earnings (Box 24)',
currency_field='currency_id',
)
# === Box 26: CPP Pensionable Earnings ===
cpp_pensionable_earnings = fields.Monetary(
string='CPP Pensionable Earnings (Box 26)',
currency_field='currency_id',
)
# === Employer portions (for reference) ===
cpp_employer = fields.Monetary(
string='CPP Employer',
currency_field='currency_id',
)
cpp2_employer = fields.Monetary(
string='CPP2 Employer',
currency_field='currency_id',
)
ei_employer = fields.Monetary(
string='EI Employer',
currency_field='currency_id',
)
# === T4 Dental Benefits Code ===
t4_dental_code = fields.Selection(
related='employee_id.t4_dental_code',
string='Dental Benefits Code',
)
# === Box 40: Other Taxable Allowances and Benefits ===
box_40_taxable_benefits = fields.Monetary(
string='Box 40: Other Taxable Allowances/Benefits',
currency_field='currency_id',
help='Taxable allowances and benefits in cash (also included in Box 14)',
)
# === Box 42: Employment Commissions ===
box_42_commissions = fields.Monetary(
string='Box 42: Employment Commissions',
currency_field='currency_id',
help='Commission earnings (also included in Box 14)',
)
# === Box 44: Union Dues ===
box_44_union_dues = fields.Monetary(
string='Box 44: Union Dues',
currency_field='currency_id',
help='Union dues deducted (pre-tax deduction, not included in Box 14)',
)
# === PDF Generation ===
filled_pdf = fields.Binary(
string='Filled PDF',
attachment=True,
)
filled_pdf_filename = fields.Char(
string='PDF Filename',
)
def _get_pdf_text_coordinates(self):
"""Get text overlay coordinates for flattened PDF
Returns dict mapping field names to (x, y, font_size, font_name) tuples
Coordinates are in points (1/72 inch), origin at bottom-left
Reads from pdf.field.position model based on template type
"""
# Determine template type from model name
template_type = None
if self._name == 'hr.t4.slip':
template_type = 'T4'
elif self._name == 'hr.t4.summary':
template_type = 'T4 Summary'
if template_type:
# Query configured positions from database
position_model = self.env['pdf.field.position']
return position_model.get_coordinates_dict(template_type)
# Fallback: return empty dict if template type not recognized
return {}
def _overlay_text_on_pdf(self, template_path, field_mapping):
"""Overlay text on a flattened PDF using reportlab
Returns base64-encoded PDF data
"""
try:
from reportlab.pdfgen import canvas
from reportlab.lib.pagesizes import letter
from PyPDF2 import PdfReader, PdfWriter
except ImportError as e:
raise UserError(f'Required library not available: {str(e)}\nPlease install reportlab and PyPDF2.')
# Get text coordinates
text_coords = self._get_pdf_text_coordinates()
if not text_coords:
# Determine template type for error message
template_type = 'T4'
if self._name == 'hr.t4.summary':
template_type = 'T4 Summary'
raise UserError(
f'Text coordinates not configured for {template_type} template. '
'Please configure PDF field positions in Payroll → Configuration → PDF Field Positions.'
)
# Read the template PDF
with open(template_path, 'rb') as template_file:
template_reader = PdfReader(template_file)
if not template_reader.pages:
raise UserError('Template PDF has no pages')
# Get first page dimensions
first_page = template_reader.pages[0]
page_width = float(first_page.mediabox.width)
page_height = float(first_page.mediabox.height)
# Create overlay PDF with text
overlay_buffer = io.BytesIO()
can = canvas.Canvas(overlay_buffer, pagesize=(page_width, page_height))
# Draw text for each field
for field_name, value in field_mapping.items():
if field_name in text_coords and value:
coord_data = text_coords[field_name]
# Handle both old format (x, y, font_size) and new format (x, y, font_size, font_name)
if len(coord_data) == 4:
x, y, font_size, font_name = coord_data
elif len(coord_data) == 3:
x, y, font_size = coord_data
font_name = 'Helvetica' # Default font
else:
continue # Skip invalid coordinate data
can.setFont(font_name, font_size)
can.drawString(x, y, str(value))
can.save()
overlay_buffer.seek(0)
# Merge overlay with template
template_file.seek(0) if hasattr(template_file, 'seek') else None
with open(template_path, 'rb') as template_file:
template_reader = PdfReader(template_file)
overlay_reader = PdfReader(overlay_buffer)
writer = PdfWriter()
for i, page in enumerate(template_reader.pages):
if i < len(overlay_reader.pages):
page.merge_page(overlay_reader.pages[i])
writer.add_page(page)
# Write to bytes
output_buffer = io.BytesIO()
writer.write(output_buffer)
return base64.b64encode(output_buffer.getvalue())
def action_fill_pdf(self):
"""Fill the T4 PDF form with data from this slip"""
self.ensure_one()
try:
# Try to import pdfrw (preferred) or PyPDF2
try:
from pdfrw import PdfReader, PdfWriter, PdfDict
use_pdfrw = True
except ImportError:
try:
import PyPDF2
use_pdfrw = False
except ImportError:
raise UserError(
'PDF library not found. Please install pdfrw or PyPDF2:\n'
'pip install pdfrw\n'
'or\n'
'pip install PyPDF2'
)
# Get PDF template path - try multiple locations
# 1. Try in static/pdf/ folder (recommended location)
module_path = os.path.dirname(os.path.dirname(__file__))
template_path = os.path.join(module_path, 'static', 'pdf', 't4-fill-25e.pdf')
# 2. Try in module root directory (fallback)
if not os.path.exists(template_path):
template_path = os.path.join(module_path, 't4-fill-25e.pdf')
# 3. Try using tools.file_path (Odoo 19)
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/static/pdf/t4-fill-25e.pdf')
except:
pass
# 4. Final fallback - root directory
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/t4-fill-25e.pdf')
except:
pass
if not os.path.exists(template_path):
raise UserError(
'T4 PDF template not found. Please ensure t4-fill-25e.pdf is in one of these locations:\n'
f'1. {os.path.join(module_path, "static", "pdf", "t4-fill-25e.pdf")} (recommended)\n'
f'2. {os.path.join(module_path, "t4-fill-25e.pdf")} (module root)\n\n'
'The system will automatically fill the PDF with data from this T4 slip when you click "Fill PDF".'
)
# Get field mapping
field_mapping = self._get_pdf_field_mapping()
# Check if we should use text overlay (for flattened PDFs)
text_coords = self._get_pdf_text_coordinates()
if text_coords:
# Use text overlay method for flattened PDF
pdf_data = self._overlay_text_on_pdf(template_path, field_mapping)
elif use_pdfrw:
# Use pdfrw to fill PDF
try:
template = PdfReader(template_path)
# Fill form fields
if hasattr(template.Root, 'AcroForm') and template.Root.AcroForm:
if hasattr(template.Root.AcroForm, 'Fields') and template.Root.AcroForm.Fields:
for field in template.Root.AcroForm.Fields:
# Get field name (can be in /T or /TU)
field_name = None
if hasattr(field, 'T'):
field_name = str(field.T).strip('()')
elif hasattr(field, 'TU'):
field_name = str(field.TU).strip('()')
if field_name and field_name in field_mapping:
value = field_mapping[field_name]
if value is not None and value != '':
# Set field value
field.V = str(value)
# Make sure field is not read-only
if hasattr(field, 'Ff'):
field.Ff = 0 # Remove read-only flag
# Write filled PDF to temporary file
import tempfile
with tempfile.NamedTemporaryFile(delete=False, suffix='.pdf') as tmp_file:
tmp_path = tmp_file.name
writer = PdfWriter()
writer.write(template, tmp_path)
# Read filled PDF
with open(tmp_path, 'rb') as f:
pdf_data = base64.b64encode(f.read())
# Clean up temp file
try:
os.remove(tmp_path)
except:
pass
except (ValueError, TypeError, Exception) as e:
# pdfrw failed to parse the PDF, fall back to PyPDF2
import logging
_logger = logging.getLogger(__name__)
_logger.warning(f'pdfrw failed to read PDF ({str(e)}), falling back to PyPDF2')
use_pdfrw = False
# Import PyPDF2 for fallback
try:
import PyPDF2
except ImportError:
raise UserError(
f'pdfrw failed to read the PDF file, and PyPDF2 is not available.\n'
f'Error: {str(e)}\n\n'
f'Please install PyPDF2: pip install PyPDF2'
)
if not use_pdfrw:
# Use PyPDF2 (fallback)
with open(template_path, 'rb') as template_file:
reader = PyPDF2.PdfReader(template_file)
writer = PyPDF2.PdfWriter()
# Copy pages
try:
for page in reader.pages:
writer.add_page(page)
except Exception as page_error:
# Check if PyCryptodome is needed
try:
from Crypto.Cipher import AES
except ImportError:
raise UserError(
f'PDF file appears to be encrypted and requires PyCryptodome for decryption.\n'
f'Error: {str(page_error)}\n\n'
f'Please install PyCryptodome: pip install pycryptodome\n'
f'Or use an unencrypted PDF template.'
)
raise
# Fill form fields (PyPDF2 approach)
if '/AcroForm' in reader.trailer['/Root']:
writer.update_page_form_field_values(writer.pages[0], field_mapping)
# Write to bytes
output_buffer = io.BytesIO()
writer.write(output_buffer)
pdf_data = base64.b64encode(output_buffer.getvalue())
# Generate filename
employee_safe = (self.employee_name or 'Employee').replace(' ', '_').replace(',', '').replace('/', '_').replace('\\', '_')[:30]
filename = f'T4_{self.tax_year}_{employee_safe}.pdf'
# Save filled PDF
self.write({
'filled_pdf': pdf_data,
'filled_pdf_filename': filename,
})
# Create attachment
attachment = self.env['ir.attachment'].create({
'name': filename,
'type': 'binary',
'datas': pdf_data,
'res_model': self._name,
'res_id': self.id,
'mimetype': 'application/pdf',
})
# Post to chatter
self.message_post(
body=f'T4 PDF generated: {filename}',
attachment_ids=[attachment.id],
)
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'PDF Generated',
'message': f'T4 PDF filled and saved: {filename}',
'type': 'success',
}
}
except Exception as e:
import traceback
error_msg = f'Error filling PDF: {str(e)}\n\nTraceback:\n{traceback.format_exc()}'
raise UserError(error_msg)
def action_extract_pdf_fields(self):
"""Helper method to extract PDF form field names (for debugging)"""
self.ensure_one()
try:
from pdfrw import PdfReader
except ImportError:
raise UserError('pdfrw library not installed. Install with: pip install pdfrw')
# Get PDF template path - try multiple locations
module_path = os.path.dirname(os.path.dirname(__file__))
template_path = os.path.join(module_path, 'static', 'pdf', 't4-fill-25e.pdf')
if not os.path.exists(template_path):
template_path = os.path.join(module_path, 't4-fill-25e.pdf')
if not os.path.exists(template_path):
try:
template_path = tools.file_path('fusion_payroll/static/pdf/t4-fill-25e.pdf')
except:
template_path = None
if not template_path or not os.path.exists(template_path):
raise UserError('T4 PDF template not found. Please ensure t4-fill-25e.pdf is in static/pdf/ or module root.')
template = PdfReader(template_path)
field_names = []
# Extract field names from all pages
for page_num, page in enumerate(template.pages, 1):
if hasattr(page, 'Annots') and page.Annots:
for annot in page.Annots:
if hasattr(annot, 'Subtype') and str(annot.Subtype) == '/Widget':
if hasattr(annot, 'T'):
field_name = str(annot.T).strip('()')
field_names.append(f'Page {page_num}: {field_name}')
# Return as message
if field_names:
message = 'PDF Form Fields Found:\n\n' + '\n'.join(field_names[:50])
if len(field_names) > 50:
message += f'\n\n... and {len(field_names) - 50} more fields'
else:
message = 'No form fields found in PDF. The PDF may not be a fillable form, or field names are stored differently.'
return {
'type': 'ir.actions.client',
'tag': 'display_notification',
'params': {
'title': 'PDF Fields',
'message': message,
'type': 'info',
'sticky': True,
}
}
def action_download_pdf(self):
"""Download the filled PDF"""
self.ensure_one()
if not self.filled_pdf:
raise UserError('No PDF has been generated yet. Please click "Fill PDF" first.')
return {
'type': 'ir.actions.act_url',
'url': f'/web/content/hr.t4.slip/{self.id}/filled_pdf/{self.filled_pdf_filename}?download=true',
'target': 'self',
}
def _get_pdf_field_mapping(self):
"""Map model fields to PDF form field names"""
# This mapping may need to be adjusted based on actual PDF form field names
# You can use action_extract_pdf_fields() to see actual field names in the PDF
mapping = {}
# Year
mapping['Year'] = str(self.tax_year)
mapping['year'] = str(self.tax_year)
mapping['YEAR'] = str(self.tax_year)
mapping['TaxYear'] = str(self.tax_year)
# Payer information (from company)
company = self.company_id
if company:
mapping['PayerName'] = company.name or ''
mapping['PayerName1'] = company.name or ''
mapping['EmployerName'] = company.name or ''
if company.street:
mapping['PayerAddress1'] = company.street
mapping['EmployerAddress1'] = company.street
if company.street2:
mapping['PayerAddress2'] = company.street2
mapping['EmployerAddress2'] = company.street2
if company.city:
city_line = company.city
if company.state_id:
city_line += f", {company.state_id.code}"
if company.zip:
city_line += f" {company.zip}"
mapping['PayerCity'] = city_line
mapping['EmployerCity'] = city_line
# Payer account number (Box 54)
settings = self.env['payroll.config.settings'].get_settings(company.id)
account_num = settings.get_cra_payroll_account_number() or company.vat or ''
mapping['PayerAccount'] = account_num
mapping['Box54'] = account_num
mapping['EmployerAccount'] = account_num
# Employee information
employee = self.employee_id
if employee:
# Employee name - format: Last name, First name
if employee.name:
name_parts = employee.name.split(',') if ',' in employee.name else employee.name.split()
if len(name_parts) >= 2:
mapping['LastName'] = name_parts[-1].strip()
mapping['FirstName'] = ' '.join(name_parts[:-1]).strip()
mapping['EmployeeLastName'] = name_parts[-1].strip()
mapping['EmployeeFirstName'] = ' '.join(name_parts[:-1]).strip()
else:
mapping['LastName'] = employee.name
mapping['EmployeeLastName'] = employee.name
# Employee address (using private_* fields in Odoo 19)
# Use getattr to safely access fields that may not exist
private_street = getattr(employee, 'private_street', None) or ''
private_street2 = getattr(employee, 'private_street2', None) or ''
private_city = getattr(employee, 'private_city', None) or ''
private_state_id = getattr(employee, 'private_state_id', None)
private_zip = getattr(employee, 'private_zip', None) or ''
if private_street or private_city:
if private_street:
mapping['EmployeeAddress1'] = private_street
if private_street2:
mapping['EmployeeAddress2'] = private_street2
if private_city:
city_line = private_city
if private_state_id:
city_line += f", {private_state_id.code}"
if private_zip:
city_line += f" {private_zip}"
mapping['EmployeeCity'] = city_line
# SIN (Box 12)
if self.sin_number:
sin_clean = self.sin_number.replace('-', '').replace(' ', '')
mapping['SIN'] = sin_clean
mapping['Box12'] = sin_clean
mapping['SocialInsuranceNumber'] = sin_clean
# Province of employment (Box 10) - from employee's work location or province
if employee and employee.address_id and employee.address_id.state_id:
mapping['Box10'] = employee.address_id.state_id.code
mapping['Province'] = employee.address_id.state_id.code
elif company and company.state_id:
mapping['Box10'] = company.state_id.code
mapping['Province'] = company.state_id.code
# Box 14: Employment Income
if self.employment_income:
mapping['Box14'] = f"{self.employment_income:.2f}"
mapping['014'] = f"{self.employment_income:.2f}"
mapping['EmploymentIncome'] = f"{self.employment_income:.2f}"
# Box 16: CPP Employee
if self.cpp_employee:
mapping['Box16'] = f"{self.cpp_employee:.2f}"
mapping['016'] = f"{self.cpp_employee:.2f}"
mapping['CPPEmployee'] = f"{self.cpp_employee:.2f}"
# Box 16A: CPP2 Employee
if self.cpp2_employee:
mapping['Box16A'] = f"{self.cpp2_employee:.2f}"
mapping['016A'] = f"{self.cpp2_employee:.2f}"
mapping['CPP2Employee'] = f"{self.cpp2_employee:.2f}"
# Box 18: EI Employee
if self.ei_employee:
mapping['Box18'] = f"{self.ei_employee:.2f}"
mapping['018'] = f"{self.ei_employee:.2f}"
mapping['EIEmployee'] = f"{self.ei_employee:.2f}"
# Box 22: Income Tax
if self.income_tax:
mapping['Box22'] = f"{self.income_tax:.2f}"
mapping['022'] = f"{self.income_tax:.2f}"
mapping['IncomeTax'] = f"{self.income_tax:.2f}"
# Box 24: EI Insurable Earnings
if self.ei_insurable_earnings:
mapping['Box24'] = f"{self.ei_insurable_earnings:.2f}"
mapping['024'] = f"{self.ei_insurable_earnings:.2f}"
mapping['EIInsurableEarnings'] = f"{self.ei_insurable_earnings:.2f}"
# Box 26: CPP Pensionable Earnings
if self.cpp_pensionable_earnings:
mapping['Box26'] = f"{self.cpp_pensionable_earnings:.2f}"
mapping['026'] = f"{self.cpp_pensionable_earnings:.2f}"
mapping['CPPPensionableEarnings'] = f"{self.cpp_pensionable_earnings:.2f}"
# Box 40: Taxable Benefits
if self.box_40_taxable_benefits:
mapping['Box40'] = f"{self.box_40_taxable_benefits:.2f}"
mapping['040'] = f"{self.box_40_taxable_benefits:.2f}"
mapping['TaxableBenefits'] = f"{self.box_40_taxable_benefits:.2f}"
# Box 42: Commissions
if self.box_42_commissions:
mapping['Box42'] = f"{self.box_42_commissions:.2f}"
mapping['042'] = f"{self.box_42_commissions:.2f}"
mapping['Commissions'] = f"{self.box_42_commissions:.2f}"
# Box 44: Union Dues
if self.box_44_union_dues:
mapping['Box44'] = f"{self.box_44_union_dues:.2f}"
mapping['044'] = f"{self.box_44_union_dues:.2f}"
mapping['UnionDues'] = f"{self.box_44_union_dues:.2f}"
# Box 45: Dental Benefits Code
if self.t4_dental_code:
mapping['Box45'] = str(self.t4_dental_code)
mapping['045'] = str(self.t4_dental_code)
mapping['DentalCode'] = str(self.t4_dental_code)
return mapping