# -*- coding: utf-8 -*-
import html as html_lib
import logging
from datetime import timedelta
from odoo import _, api, fields, models
from odoo.exceptions import UserError
from odoo.tools import html_sanitize, is_html_empty, plaintext2html
_logger = logging.getLogger(__name__)
class MyCompanyPost(models.Model):
_name = 'mycompany.post'
_description = 'Company Feed Post'
_inherit = ['mail.thread', 'mail.activity.mixin']
_order = 'is_pinned desc, create_date desc'
name = fields.Char(string='Title', required=True, tracking=True)
body = fields.Html(string='Content', sanitize=True)
post_type = fields.Selection(
selection=[
('post', 'Post'),
('announcement', 'Announcement'),
('kudos', 'Kudos'),
],
default='post',
required=True,
tracking=True,
)
author_id = fields.Many2one(
comodel_name='res.users',
string='Author',
default=lambda self: self.env.user,
required=True,
index=True,
)
author_partner_id = fields.Many2one(
related='author_id.partner_id',
store=True,
)
partner_ids = fields.Many2many(
comodel_name='res.partner',
relation='mycompany_post_partner_rel',
column1='post_id',
column2='partner_id',
string='Tagged People',
)
department_ids = fields.Many2many(
comodel_name='hr.department',
relation='mycompany_post_department_rel',
column1='post_id',
column2='department_id',
string='Teams / Departments',
)
kudos_recipient_id = fields.Many2one(
comodel_name='res.users',
string='Kudos For',
tracking=True,
)
attachment_ids = fields.Many2many(
comodel_name='ir.attachment',
relation='mycompany_post_attachment_rel',
column1='post_id',
column2='attachment_id',
string='Images',
)
image_count = fields.Integer(compute='_compute_image_count')
comment_count = fields.Integer(compute='_compute_comment_count')
company_id = fields.Many2one(
comodel_name='res.company',
default=lambda self: self.env.company,
required=True,
index=True,
)
is_pinned = fields.Boolean(
string='Pinned',
default=False,
help='Pinned announcements stay at the top of the feed.',
)
active = fields.Boolean(default=True)
@api.depends('attachment_ids')
def _compute_image_count(self):
for post in self:
post.image_count = len(post.attachment_ids)
@api.depends('message_ids')
def _compute_comment_count(self):
for post in self:
post.comment_count = len(post.message_ids.filtered(
lambda m: m.message_type == 'comment'
))
@api.constrains('post_type', 'kudos_recipient_id')
def _check_kudos_recipient(self):
for post in self:
if post.post_type == 'kudos' and not post.kudos_recipient_id:
raise UserError(_('Kudos posts must specify who the kudos is for.'))
@api.model_create_multi
def create(self, vals_list):
posts = super().create(vals_list)
posts._notify_tagged_and_teams()
return posts
def write(self, vals):
res = super().write(vals)
if any(key in vals for key in ('partner_ids', 'department_ids', 'post_type')):
self._notify_tagged_and_teams()
return res
def _notify_tagged_and_teams(self):
"""Notify tagged partners and members of targeted departments."""
for post in self:
partners = post.partner_ids
if post.kudos_recipient_id:
partners |= post.kudos_recipient_id.partner_id
for department in post.department_ids:
partners |= department.member_ids.mapped('user_id.partner_id')
partners -= post.author_id.partner_id
if not partners:
continue
subtype = self.env.ref('mail.mt_comment', raise_if_not_found=False)
post.message_post(
body=_('You were mentioned in a company update.'),
partner_ids=partners.ids,
subtype_id=subtype.id if subtype else False,
message_type='notification',
)
# ------------------------------------------------------------------
# Hub API — called from the OWL client action
# ------------------------------------------------------------------
@api.model
def _format_hub_body(self, body):
"""Return safe HTML for hub rendering (fixes double-escaped mail bodies)."""
if not body or is_html_empty(body):
return ''
text = body
if '<' in text and '>' in text:
text = html_lib.unescape(text)
return html_sanitize(text)
@api.model
def get_hub_settings(self):
"""Return hub branding settings for the OWL client action."""
name = self.env['ir.config_parameter'].sudo().get_param(
'my_company.hub_name', 'My Company'
)
name = (name or 'My Company').strip() or 'My Company'
return {'hub_name': name}
@api.model
def get_hub_feed(self, limit=30, offset=0):
"""Return serialised feed posts for the social hub."""
domain = [
('company_id', '=', self.env.company.id),
('active', '=', True),
]
posts = self.search(domain, limit=limit, offset=offset)
return [post._prepare_hub_data() for post in posts]
def _prepare_hub_data(self):
self.ensure_one()
return {
'id': self.id,
'name': self.name,
'body': self._format_hub_body(self.body or ''),
'post_type': self.post_type,
'author_id': self.author_id.id,
'author_name': self.author_id.name,
'author_avatar': f'/web/image/res.users/{self.author_id.id}/avatar_128',
'kudos_recipient_id': self.kudos_recipient_id.id or False,
'kudos_recipient_name': self.kudos_recipient_id.name or '',
'is_pinned': self.is_pinned,
'create_date': fields.Datetime.to_string(self.create_date),
'comment_count': self.comment_count,
'tagged_partner_ids': self.partner_ids.ids,
'tagged_partner_names': self.partner_ids.mapped('name'),
'department_names': self.department_ids.mapped('name'),
'images': [{
'id': att.id,
'name': att.name,
'url': f'/web/image/ir.attachment/{att.id}/datas',
} for att in self.attachment_ids],
'comments': self._prepare_hub_comments(),
}
def _prepare_hub_comments(self, limit=10):
self.ensure_one()
comments = self.message_ids.filtered(
lambda m: m.message_type == 'comment'
).sorted(key=lambda m: m.date, reverse=True)[:limit]
result = []
for msg in reversed(comments):
result.append({
'id': msg.id,
'body': self._format_hub_body(msg.body or ''),
'author_name': msg.author_id.name if msg.author_id else _('Unknown'),
'author_avatar': (
f'/web/image/res.partner/{msg.author_id.id}/avatar_128'
if msg.author_id else ''
),
'date': fields.Datetime.to_string(msg.date),
})
return result
@api.model
def create_hub_post(self, vals):
"""Create a post from the hub composer."""
post_vals = {
'name': vals.get('name') or _('Update'),
'body': plaintext2html((vals.get('body') or '').strip()) if vals.get('body') else '',
'post_type': vals.get('post_type', 'post'),
'company_id': self.env.company.id,
}
if vals.get('kudos_recipient_id'):
post_vals['kudos_recipient_id'] = vals['kudos_recipient_id']
if vals.get('partner_ids'):
post_vals['partner_ids'] = [(6, 0, vals['partner_ids'])]
if vals.get('department_ids'):
post_vals['department_ids'] = [(6, 0, vals['department_ids'])]
post = self.create(post_vals)
# Link pre-uploaded attachments
attachment_ids = vals.get('attachment_ids') or []
if attachment_ids:
attachments = self.env['ir.attachment'].browse(attachment_ids)
attachments.write({'res_model': self._name, 'res_id': post.id})
post.attachment_ids = [(6, 0, attachment_ids)]
return post._prepare_hub_data()
def add_hub_comment(self, body):
"""Post a comment on a feed item from the hub."""
self.ensure_one()
if not body or not body.strip():
raise UserError(_('Comment cannot be empty.'))
msg = self.message_post(
body=plaintext2html(body.strip()),
message_type='comment',
subtype_xmlid='mail.mt_comment',
)
return {
'id': msg.id,
'body': self._format_hub_body(msg.body or ''),
'author_name': self.env.user.name,
'author_avatar': f'/web/image/res.users/{self.env.user.id}/avatar_128',
'date': fields.Datetime.to_string(msg.date),
}
@api.model
def get_hub_users(self):
"""Return internal users with online status for the chat panel."""
users = self.env['res.users'].search([
('share', '=', False),
('active', '=', True),
('company_ids', 'in', self.env.company.id),
], order='name')
EmployeePublic = self.env['hr.employee.public']
result = []
for user in users:
if user.id == self.env.user.id:
continue
public_employee = EmployeePublic.search([('user_id', '=', user.id)], limit=1)
result.append({
'id': user.id,
'name': user.name,
'partner_id': user.partner_id.id,
'avatar': f'/web/image/res.users/{user.id}/avatar_128',
'im_status': user.im_status or 'offline',
'job_title': public_employee.job_title if public_employee else '',
})
return result
@api.model
def get_hub_birthdays(self):
"""Return colleagues with birthdays today or within the next 7 days."""
today = fields.Date.context_today(self)
employees = self.env['hr.employee'].sudo().search([
('company_id', 'in', self.env.company.ids),
('birthday', '!=', False),
('user_id', '!=', False),
('user_id.active', '=', True),
])
birthdays = []
for employee in employees:
if employee.user_id.id == self.env.user.id:
continue
this_year = employee.birthday.replace(year=today.year)
if this_year < today:
this_year = employee.birthday.replace(year=today.year + 1)
days_until = (this_year - today).days
if days_until > 7:
continue
if days_until == 0:
label = _('Today')
elif days_until == 1:
label = _('Tomorrow')
else:
label = _('In %s days') % days_until
birthdays.append({
'name': employee.name,
'user_id': employee.user_id.id,
'avatar': f'/web/image/hr.employee/{employee.id}/avatar_128',
'label': label,
'is_today': days_until == 0,
'sort_key': days_until,
})
birthdays.sort(key=lambda item: item['sort_key'])
return [{
'name': item['name'],
'user_id': item['user_id'],
'avatar': item['avatar'],
'label': item['label'],
'is_today': item['is_today'],
} for item in birthdays[:12]]
@api.model
def _cron_engagement_reminder(self):
"""Weekly nudge for users who have not posted recently."""
if not self.env['ir.config_parameter'].sudo().get_param(
'my_company.engagement_reminders', 'True'
) == 'True':
return
Post = self.env['mycompany.post'].sudo()
week_ago = fields.Datetime.now() - timedelta(days=7)
users = self.env['res.users'].sudo().search([
('share', '=', False),
('active', '=', True),
])
for user in users:
recent = Post.search_count([
('author_id', '=', user.id),
('create_date', '>=', week_ago),
])
if recent:
continue
user.partner_id.message_post(
body=_(
'Share something with the team this week — '
'a kudos, update, or photo on My Company!'
),
subject=_('My Company — stay connected'),
message_type='notification',
subtype_xmlid='mail.mt_note',
)
_logger.info('My Company engagement reminders sent.')