Files
Odoo-Modules/fusion_plating/fusion_plating_portal/controllers/portal.py
gsinghpal c2590a99ff feat(portal): welcome-line summary counts on /my/home + tests
Adds active_job_count, awaiting_review_count, ready_to_ship_count
to the dashboard context. Tests verify partition is correct across
the fp.portal.job and fp.quote.request state machines.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-17 02:41:17 -04:00

875 lines
31 KiB
Python

# -*- coding: utf-8 -*-
# Copyright 2026 Nexa Systems Inc.
# License OPL-1 (Odoo Proprietary License v1.0)
# Part of the Fusion Plating product family.
import base64
import json
from odoo import _, http
from odoo.exceptions import AccessError, MissingError
from odoo.http import request
from odoo.addons.portal.controllers.portal import (
CustomerPortal,
pager as portal_pager,
)
class FpCustomerPortal(CustomerPortal):
"""Customer portal extension for Fusion Plating.
Adds a rich dashboard, Quote Requests with part lines, Jobs with
segmented progress, Purchase Orders, Invoices, Shipping, and
Certifications sections.
"""
# ==========================================================================
# Portal Home -- counters
# ==========================================================================
def _prepare_home_portal_values(self, counters):
values = super()._prepare_home_portal_values(counters)
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
if 'fp_quote_request_count' in counters:
values['fp_quote_request_count'] = request.env[
'fusion.plating.quote.request'
].search_count([('partner_id', 'child_of', commercial.id)])
if 'fp_portal_job_count' in counters:
values['fp_portal_job_count'] = request.env[
'fusion.plating.portal.job'
].search_count([('partner_id', 'child_of', commercial.id)])
if 'fp_purchase_order_count' in counters:
values['fp_purchase_order_count'] = request.env[
'sale.order'
].sudo().search_count([
('partner_id', 'child_of', commercial.id),
('state', '=', 'sale'),
])
if 'fp_invoice_count' in counters:
values['fp_invoice_count'] = request.env[
'account.move'
].sudo().search_count([
('partner_id', 'child_of', commercial.id),
('move_type', '=', 'out_invoice'),
('state', '=', 'posted'),
])
if 'fp_delivery_count' in counters:
values['fp_delivery_count'] = request.env[
'stock.picking'
].sudo().search_count([
('partner_id', 'child_of', commercial.id),
('picking_type_code', '=', 'outgoing'),
('state', '=', 'done'),
])
if 'fp_certification_count' in counters:
values['fp_certification_count'] = request.env[
'fusion.plating.portal.job'
].search_count([
('partner_id', 'child_of', commercial.id),
('coc_attachment_id', '!=', False),
])
return values
# ==========================================================================
# Helpers
# ==========================================================================
def _fp_quote_request_get_page_view_values(self, quote_request, access_token, **kwargs):
values = {
'page_name': 'fp_quote_request',
'quote_request': quote_request,
}
return self._get_page_view_values(
quote_request,
access_token,
values,
'my_fp_quote_requests_history',
False,
**kwargs,
)
def _fp_portal_job_get_page_view_values(self, job, access_token, **kwargs):
values = {
'page_name': 'fp_portal_job',
'job': job,
}
return self._get_page_view_values(
job,
access_token,
values,
'my_fp_jobs_history',
False,
**kwargs,
)
def _fp_get_partner_domain(self):
"""Return base domain filtering to commercial partner tree."""
partner = request.env.user.partner_id
return partner.commercial_partner_id
# ==========================================================================
# DASHBOARD
# ==========================================================================
@http.route(
['/my/home'],
type='http',
auth='user',
website=True,
)
def home(self, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
# Recent quotes (5)
Quote = request.env['fusion.plating.quote.request']
quote_domain = [('partner_id', 'child_of', commercial.id)]
recent_quotes = Quote.search(
quote_domain, order='create_date desc', limit=5
)
quote_count = Quote.search_count(quote_domain)
# Recent POs (5) -- sale.order with state=sale
SO = request.env['sale.order'].sudo()
po_domain = [
('partner_id', 'child_of', commercial.id),
('state', '=', 'sale'),
]
recent_pos = SO.search(po_domain, order='date_order desc', limit=5)
po_count = SO.search_count(po_domain)
# Recent jobs (5)
Job = request.env['fusion.plating.portal.job']
job_domain = [('partner_id', 'child_of', commercial.id)]
recent_jobs = Job.search(
job_domain, order='received_date desc, id desc', limit=5
)
job_count = Job.search_count(job_domain)
# Certifications (jobs with CoC)
cert_domain = [
('partner_id', 'child_of', commercial.id),
('coc_attachment_id', '!=', False),
]
recent_certs = Job.search(
cert_domain, order='actual_ship_date desc, id desc', limit=5
)
cert_count = Job.search_count(cert_domain)
# Deliveries (stock.picking outgoing done)
Picking = request.env['stock.picking'].sudo()
delivery_domain = [
('partner_id', 'child_of', commercial.id),
('picking_type_code', '=', 'outgoing'),
('state', '=', 'done'),
]
recent_deliveries = Picking.search(
delivery_domain, order='date_done desc', limit=5
)
delivery_count = Picking.search_count(delivery_domain)
# Invoices
Invoice = request.env['account.move'].sudo()
invoice_domain = [
('partner_id', 'child_of', commercial.id),
('move_type', '=', 'out_invoice'),
('state', '=', 'posted'),
]
recent_invoices = Invoice.search(
invoice_domain, order='invoice_date desc', limit=5
)
invoice_count = Invoice.search_count(invoice_domain)
# Welcome-line summary counts (jobs-forward dashboard).
active_job_count = Job.search_count([
('partner_id', 'child_of', commercial.id),
('state', 'in', ['received', 'in_progress', 'quality_check']),
])
awaiting_review_count = Quote.search_count([
('partner_id', 'child_of', commercial.id),
('state', '=', 'quoted'),
])
ready_to_ship_count = Job.search_count([
('partner_id', 'child_of', commercial.id),
('state', '=', 'ready_to_ship'),
])
values = {
'page_name': 'fp_dashboard',
'partner': partner,
# Quotes
'recent_quotes': recent_quotes,
'quote_count': quote_count,
# Purchase Orders
'recent_pos': recent_pos,
'po_count': po_count,
# Jobs / Parts Portal
'recent_jobs': recent_jobs,
'job_count': job_count,
# Certifications
'recent_certs': recent_certs,
'cert_count': cert_count,
# Deliveries
'recent_deliveries': recent_deliveries,
'delivery_count': delivery_count,
# Invoices
'recent_invoices': recent_invoices,
'invoice_count': invoice_count,
# Welcome-line summary
'active_job_count': active_job_count,
'awaiting_review_count': awaiting_review_count,
'ready_to_ship_count': ready_to_ship_count,
}
return request.render(
'fusion_plating_portal.fp_portal_home_dashboard',
values,
)
# ==========================================================================
# QUOTE REQUESTS -- list with tabs
# ==========================================================================
@http.route(
['/my/quote_requests', '/my/quote_requests/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_quote_requests(self, page=1, sortby=None, filterby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
Quote = request.env['fusion.plating.quote.request']
domain = [('partner_id', 'child_of', commercial.id)]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'create_date desc'},
'name': {'label': _('Reference'), 'order': 'name desc'},
'state': {'label': _('Status'), 'order': 'state'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
# Tab filters
searchbar_filters = {
'all': {'label': _('All'), 'domain': []},
'active': {'label': _('Active'), 'domain': [
('state', 'in', ['new', 'under_review', 'quoted']),
]},
'converted': {'label': _('Converted'), 'domain': [
('state', '=', 'accepted'),
]},
'declined': {'label': _('Declined'), 'domain': [
('state', 'in', ['declined', 'expired']),
]},
}
if not filterby or filterby not in searchbar_filters:
filterby = 'all'
domain += searchbar_filters[filterby]['domain']
total = Quote.search_count(domain)
pager = portal_pager(
url='/my/quote_requests',
url_args={'sortby': sortby, 'filterby': filterby},
total=total,
page=page,
step=self._items_per_page,
)
quote_requests = Quote.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
request.session['my_fp_quote_requests_history'] = quote_requests.ids[:100]
values = {
'quote_requests': quote_requests,
'page_name': 'fp_quote_requests',
'pager': pager,
'default_url': '/my/quote_requests',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
'searchbar_filters': searchbar_filters,
'filterby': filterby,
}
return request.render(
'fusion_plating_portal.portal_my_quote_requests',
values,
)
# ==========================================================================
# QUOTE REQUESTS -- detail
# ==========================================================================
@http.route(
['/my/quote_requests/<int:request_id>'],
type='http',
auth='user',
website=True,
)
def portal_my_quote_request(self, request_id, access_token=None, **kw):
try:
quote_sudo = self._document_check_access(
'fusion.plating.quote.request',
request_id,
access_token,
)
except (AccessError, MissingError):
return request.redirect('/my')
values = self._fp_quote_request_get_page_view_values(
quote_sudo, access_token, **kw
)
return request.render(
'fusion_plating_portal.portal_my_quote_request',
values,
)
# ==========================================================================
# QUOTE REQUESTS -- new form (enhanced)
# ==========================================================================
@http.route(
['/my/quote_requests/new'],
type='http',
auth='user',
website=True,
)
def portal_new_quote_request(self, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
process_types = request.env['fusion.plating.process.type'].sudo().search(
[('active', '=', True)]
)
# Shipping addresses: child contacts of type 'delivery' or 'other'
addresses = request.env['res.partner'].sudo().search([
('parent_id', '=', commercial.id),
('type', 'in', ['delivery', 'other', 'contact']),
])
# Products available for this customer (all saleable products)
products = request.env['product.product'].sudo().search([
('sale_ok', '=', True),
('active', '=', True),
], limit=200, order='default_code, name')
values = {
'page_name': 'fp_quote_request_new',
'process_types': process_types,
'partner': partner,
'commercial': commercial,
'addresses': addresses,
'products': products,
'error': kw.get('error'),
'form_data': kw,
}
return request.render(
'fusion_plating_portal.portal_new_quote_request_form',
values,
)
# ==========================================================================
# QUOTE REQUESTS -- submit (enhanced for multi-part)
# ==========================================================================
@http.route(
['/my/quote_requests/submit'],
type='http',
auth='user',
website=True,
methods=['POST'],
csrf=True,
)
def portal_submit_quote_request(self, **post):
partner = request.env.user.partner_id
# Basic validation: need at least one part description or a general description
parts_json = post.get('parts_data', '[]')
try:
parts_data = json.loads(parts_json)
except (json.JSONDecodeError, TypeError):
parts_data = []
if not parts_data and not post.get('part_description'):
return request.redirect(
'/my/quote_requests/new?error=missing_description'
)
process_type_ids = []
raw_processes = request.httprequest.form.getlist('process_type_ids')
for pt in raw_processes:
try:
process_type_ids.append(int(pt))
except (TypeError, ValueError):
continue
try:
quantity = int(post.get('quantity') or 0)
except (TypeError, ValueError):
quantity = 0
vals = {
'partner_id': partner.id,
'contact_name': post.get('contact_name') or partner.name,
'contact_email': post.get('contact_email') or partner.email,
'contact_phone': post.get('contact_phone') or partner.phone,
'company_name': post.get('company_name')
or (partner.parent_id.name if partner.parent_id else partner.name),
'part_description': post.get('part_description'),
'quantity': quantity,
'special_instructions': post.get('special_instructions'),
'state': 'new',
'billing_same_as_shipping': bool(post.get('billing_same_as_shipping')),
}
# Addresses
if post.get('shipping_address_id'):
try:
vals['shipping_address_id'] = int(post['shipping_address_id'])
except (TypeError, ValueError):
pass
if post.get('billing_address_id') and not post.get('billing_same_as_shipping'):
try:
vals['billing_address_id'] = int(post['billing_address_id'])
except (TypeError, ValueError):
pass
if post.get('target_delivery'):
vals['target_delivery'] = post.get('target_delivery')
if process_type_ids:
vals['process_type_ids'] = [(6, 0, process_type_ids)]
quote = request.env['fusion.plating.quote.request'].sudo().create(vals)
# Create part lines from JSON data
if parts_data:
Line = request.env['fusion.plating.quote.request.line'].sudo()
for idx, part in enumerate(parts_data):
line_vals = {
'request_id': quote.id,
'sequence': (idx + 1) * 10,
'part_number': part.get('part_number', ''),
'quantity': int(part.get('quantity', 1) or 1),
'count': int(part.get('count', 1) or 1),
'description': part.get('description', ''),
'spec_text': part.get('spec_text', ''),
}
if part.get('product_id'):
try:
line_vals['product_id'] = int(part['product_id'])
except (TypeError, ValueError):
pass
Line.create(line_vals)
# Handle file uploads (general attachments on the quote)
files = request.httprequest.files.getlist('drawing_attachments')
attachment_ids = []
for f in files:
if not f or not f.filename:
continue
try:
content = f.read()
if not content:
continue
attachment = request.env['ir.attachment'].sudo().create({
'name': f.filename,
'datas': base64.b64encode(content),
'res_model': 'fusion.plating.quote.request',
'res_id': quote.id,
'type': 'binary',
})
attachment_ids.append(attachment.id)
except Exception:
continue
if attachment_ids:
quote.sudo().write({
'drawing_attachment_ids': [(6, 0, attachment_ids)],
})
# Handle per-line file uploads (line_file_0, line_file_1, ...)
for idx in range(len(parts_data)):
line_files = request.httprequest.files.getlist('line_file_%d' % idx)
line_att_ids = []
for f in line_files:
if not f or not f.filename:
continue
try:
content = f.read()
if not content:
continue
attachment = request.env['ir.attachment'].sudo().create({
'name': f.filename,
'datas': base64.b64encode(content),
'res_model': 'fusion.plating.quote.request.line',
'res_id': quote.line_ids[idx].id if idx < len(quote.line_ids) else 0,
'type': 'binary',
})
line_att_ids.append(attachment.id)
except Exception:
continue
if line_att_ids and idx < len(quote.line_ids):
quote.line_ids[idx].sudo().write({
'attachment_ids': [(6, 0, line_att_ids)],
})
return request.redirect('/my/quote_requests/%s' % quote.id)
# ==========================================================================
# JOBS -- list
# ==========================================================================
@http.route(
['/my/jobs', '/my/jobs/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_jobs(self, page=1, sortby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
Job = request.env['fusion.plating.portal.job']
domain = [('partner_id', 'child_of', commercial.id)]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'received_date desc, id desc'},
'name': {'label': _('Reference'), 'order': 'name desc'},
'state': {'label': _('Status'), 'order': 'state'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
total = Job.search_count(domain)
pager = portal_pager(
url='/my/jobs',
url_args={'sortby': sortby},
total=total,
page=page,
step=self._items_per_page,
)
jobs = Job.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
request.session['my_fp_jobs_history'] = jobs.ids[:100]
values = {
'jobs': jobs,
'page_name': 'fp_jobs',
'pager': pager,
'default_url': '/my/jobs',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
}
return request.render(
'fusion_plating_portal.portal_my_jobs',
values,
)
# ==========================================================================
# JOBS -- detail
# ==========================================================================
@http.route(
['/my/jobs/<int:job_id>'],
type='http',
auth='user',
website=True,
)
def portal_my_job(self, job_id, access_token=None, **kw):
try:
job_sudo = self._document_check_access(
'fusion.plating.portal.job',
job_id,
access_token,
)
except (AccessError, MissingError):
return request.redirect('/my')
values = self._fp_portal_job_get_page_view_values(
job_sudo, access_token, **kw
)
values['progress_percent'] = job_sudo._progress_percent()
return request.render(
'fusion_plating_portal.portal_my_job',
values,
)
# ==========================================================================
# JOBS -- download CoC
# ==========================================================================
@http.route(
['/my/jobs/<int:job_id>/coc'],
type='http',
auth='user',
website=True,
)
def portal_download_coc(self, job_id, access_token=None, **kw):
try:
job_sudo = self._document_check_access(
'fusion.plating.portal.job',
job_id,
access_token,
)
except (AccessError, MissingError):
return request.redirect('/my')
if not job_sudo.coc_attachment_id:
return request.redirect('/my/jobs/%s' % job_id)
attachment = job_sudo.coc_attachment_id.sudo()
return request.env['ir.binary']._get_stream_from(
attachment, 'raw'
).get_response(as_attachment=True)
# ==========================================================================
# PURCHASE ORDERS -- list
# ==========================================================================
@http.route(
['/my/purchase_orders', '/my/purchase_orders/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_purchase_orders(self, page=1, sortby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
SO = request.env['sale.order'].sudo()
domain = [
('partner_id', 'child_of', commercial.id),
('state', '=', 'sale'),
]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'date_order desc'},
'name': {'label': _('Reference'), 'order': 'name desc'},
'amount': {'label': _('Amount'), 'order': 'amount_total desc'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
total = SO.search_count(domain)
pager = portal_pager(
url='/my/purchase_orders',
url_args={'sortby': sortby},
total=total,
page=page,
step=self._items_per_page,
)
orders = SO.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
values = {
'orders': orders,
'page_name': 'fp_purchase_orders',
'pager': pager,
'default_url': '/my/purchase_orders',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
}
return request.render(
'fusion_plating_portal.portal_my_purchase_orders',
values,
)
# ==========================================================================
# INVOICES -- list
# ==========================================================================
@http.route(
['/my/fp_invoices', '/my/fp_invoices/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_fp_invoices(self, page=1, sortby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
Invoice = request.env['account.move'].sudo()
domain = [
('partner_id', 'child_of', commercial.id),
('move_type', '=', 'out_invoice'),
('state', '=', 'posted'),
]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'invoice_date desc'},
'name': {'label': _('Reference'), 'order': 'name desc'},
'amount': {'label': _('Amount'), 'order': 'amount_total desc'},
'due': {'label': _('Due Date'), 'order': 'invoice_date_due asc'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
total = Invoice.search_count(domain)
pager = portal_pager(
url='/my/fp_invoices',
url_args={'sortby': sortby},
total=total,
page=page,
step=self._items_per_page,
)
invoices = Invoice.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
values = {
'invoices': invoices,
'page_name': 'fp_invoices',
'pager': pager,
'default_url': '/my/fp_invoices',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
}
return request.render(
'fusion_plating_portal.portal_my_fp_invoices',
values,
)
# ==========================================================================
# SHIPPING / DELIVERIES -- list
# ==========================================================================
@http.route(
['/my/deliveries', '/my/deliveries/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_deliveries(self, page=1, sortby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
Picking = request.env['stock.picking'].sudo()
domain = [
('partner_id', 'child_of', commercial.id),
('picking_type_code', '=', 'outgoing'),
('state', '=', 'done'),
]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'date_done desc'},
'name': {'label': _('Reference'), 'order': 'name desc'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
total = Picking.search_count(domain)
pager = portal_pager(
url='/my/deliveries',
url_args={'sortby': sortby},
total=total,
page=page,
step=self._items_per_page,
)
deliveries = Picking.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
values = {
'deliveries': deliveries,
'page_name': 'fp_deliveries',
'pager': pager,
'default_url': '/my/deliveries',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
}
return request.render(
'fusion_plating_portal.portal_my_deliveries',
values,
)
# ==========================================================================
# CERTIFICATIONS -- list
# ==========================================================================
@http.route(
['/my/certifications', '/my/certifications/page/<int:page>'],
type='http',
auth='user',
website=True,
)
def portal_my_certifications(self, page=1, sortby=None, **kw):
partner = request.env.user.partner_id
commercial = partner.commercial_partner_id
Job = request.env['fusion.plating.portal.job']
domain = [
('partner_id', 'child_of', commercial.id),
('coc_attachment_id', '!=', False),
]
searchbar_sortings = {
'date': {'label': _('Newest'), 'order': 'actual_ship_date desc, id desc'},
'name': {'label': _('Job Reference'), 'order': 'name desc'},
}
if not sortby:
sortby = 'date'
order = searchbar_sortings[sortby]['order']
total = Job.search_count(domain)
pager = portal_pager(
url='/my/certifications',
url_args={'sortby': sortby},
total=total,
page=page,
step=self._items_per_page,
)
cert_jobs = Job.search(
domain,
order=order,
limit=self._items_per_page,
offset=pager['offset'],
)
values = {
'cert_jobs': cert_jobs,
'page_name': 'fp_certifications',
'pager': pager,
'default_url': '/my/certifications',
'searchbar_sortings': searchbar_sortings,
'sortby': sortby,
}
return request.render(
'fusion_plating_portal.portal_my_certifications',
values,
)
# ==========================================================================
# JSONRPC -- product search for RFQ form
# ==========================================================================
@http.route(
['/my/quote_requests/search_products'],
type='jsonrpc',
auth='user',
website=True,
)
def portal_search_products(self, query='', **kw):
"""Search products for the RFQ part number dropdown."""
products = request.env['product.product'].sudo().search([
('sale_ok', '=', True),
('active', '=', True),
'|', '|',
('name', 'ilike', query),
('default_code', 'ilike', query),
('barcode', 'ilike', query),
], limit=20, order='default_code, name')
return [{
'id': p.id,
'name': p.name,
'default_code': p.default_code or '',
'display': '[%s] %s' % (p.default_code, p.name) if p.default_code else p.name,
} for p in products]