第15讲、Odoo 18 中自动任务(ir.cron) 的实现原理与应用-海口c网

引言

在企业级应用中，自动化是提高效率的关键因素。Odoo 作为一款强大的企业资源规划（ERP）系统，提供了完善的自动化工具，其中 ir.cron 模型是实现定时任务的核心组件。本文将深入探讨 Odoo 18 中 ir.cron 的实现原理、运行机制、应用场景以及详细的使用案例，帮助开发者和管理员更好地利用这一功能。

ir.cron 实现原理

ir.cron 是 Odoo 中负责定时任务调度的核心模型，它通过在数据库中创建记录来定义需要定期执行的任务。当 Odoo 服务启动时，系统会初始化一个任务调度器，该调度器会定期检查 ir.cron 表中的记录，并根据配置的时间参数执行相应的任务。

基本原理

Odoo 使后台作业运行变得简单：只需在 ir.cron 表中插入一条记录，Odoo 就会按照定义执行它。每条 ir.cron 记录包含了任务的执行时间、执行间隔、执行方法等信息，系统会根据这些信息来调度任务的执行。

数据结构

ir.cron 模型的主要字段包括：

name：定时任务的名称，主要用于日志显示和用户界面
model_id：关联到要执行方法的模型
state：执行类型，通常为 “code”，表示执行 Python 代码
code：要执行的 Python 代码或方法
user_id：执行任务的用户 ID，通常是系统管理员
interval_number：执行间隔的数值
interval_type：执行间隔的单位（分钟、小时、天、周、月）
numbercall：执行次数，-1 表示无限次
doall：服务器重启时是否执行错过的任务
nextcall：下次执行的时间
priority：优先级，0-10，数字越小优先级越高
active：是否激活

image.png

配置方式

在 Odoo 18 中，可以通过两种方式配置 ir.cron：

通过用户界面：在 “设置” > “技术” > “自动化” > “计划动作” 中创建和管理
通过代码：在模块的 data 目录下创建 XML 文件，定义 ir.cron 记录

运行机制

在 Odoo 中，ir.cron 作为主进程的一部分运行，而不是独立的进程或线程。具体来说：

进程关系：ir.cron 任务在 Odoo 主进程内执行，不会创建新的进程。Odoo 服务启动时，会在主进程中初始化一个任务调度器。
线程实现：在 Odoo 的实现中，cron 任务调度器实际上是在主进程的一个专用线程中运行的。该线程负责定期检查 ir.cron 表中的记录，并在适当的时间触发任务执行。
资源共享：由于 cron 任务在主进程内执行，它们会共享 Odoo 主进程的资源，包括内存、数据库连接池等。这也是为什么长时间运行的 cron 任务可能会影响系统整体性能的原因。
事务处理：每个 cron 任务执行时会创建自己的数据库事务，任务完成后会自动提交，若发生错误则会回滚，确保数据一致性。
锁机制：为防止在多 Odoo 实例部署环境中重复执行任务，Odoo 使用了数据库级别的锁机制。只有获取到锁的进程才能执行对应的定时任务，保证任务不会被多次并发执行。

这种设计使得 cron 任务能够直接访问 Odoo 的 ORM 系统和业务逻辑，极大地方便了自动化开发。但也意味着需要谨慎设计任务，避免对主系统性能造成负面影响，尤其是避免在 cron 任务中执行耗时操作。

调度流程

初始化：Odoo 服务启动时，会初始化一个任务调度器
任务扫描：调度器定期扫描 ir.cron 表中的记录
时间检查：对于每条记录，检查 nextcall 字段是否到达执行时间
任务执行：如果到达执行时间，系统会以指定用户的身份执行任务
状态更新：执行完成后，更新 nextcall 字段为下次执行时间
执行计数：如果 numbercall 不是 -1，则减少一次计数

执行环境

ir.cron 任务在 Odoo 服务器进程中执行，而不是在单独的进程中。这意味着：

任务执行时会共享 Odoo 服务器的资源
长时间运行的任务可能会影响系统性能
任务执行时会创建新的数据库事务
任务执行完成后会自动提交事务

错误处理

当 ir.cron 任务执行失败时：

错误会被记录在服务器日志中
任务不会自动重试
如果设置了 doall=True，服务器重启后会尝试执行错过的任务

锁机制

为了防止多个 Odoo 实例同时执行同一个定时任务，ir.cron 使用了数据库级别的锁机制：

执行任务前，系统会尝试获取一个命名锁
如果锁已被其他进程持有，当前进程会跳过该任务
这确保了在多实例部署环境中，每个任务只会被执行一次

应用场景

ir.cron 在 Odoo 中有广泛的应用场景，主要包括：

1. 定期邮件通知

每日/每周销售报表发送
客户生日祝福邮件
订阅内容更新通知
账单到期提醒

2. 自动化数据清理

清理过期的临时数据
归档旧的系统日志
删除过期的附件文件
压缩历史数据

3. 业务状态自动更新

订单超时自动取消
库存预警自动通知
客户信用额度自动更新
合同到期状态变更

4. 系统健康检查

数据库备份验证
系统性能监控
接口连接状态检查
数据一致性验证

使用案例

下面将通过几个具体的案例，详细说明如何在 Odoo 18 中使用 ir.cron 实现各种自动化任务。

案例一：定期发送邮件通知

需求描述

每周一早上 9 点自动向销售团队发送上周的销售统计报告。

实现步骤

创建一个自定义模块 sales_weekly_report
在模块中定义发送报告的方法
配置 ir.cron 定时任务

代码示例

1. 模型定义 (models/sales_report.py)

from odoo import models, fields, api
from datetime import datetime, timedeltaclass SalesReport(models.Model):_name = 'sales.report'_description = '销售周报'@api.modeldef send_weekly_report(self):"""发送每周销售报告"""# 计算上周的日期范围today = fields.Date.today()weekday = today.weekday()date_from = today - timedelta(days=weekday+7)date_to = today - timedelta(days=weekday+1)# 查询上周的销售订单sales_orders = self.env['sale.order'].search([('date_order', '>=', date_from),('date_order', '<=', date_to),('state', 'in', ['sale', 'done'])])# 计算销售统计数据total_amount = sum(order.amount_total for order in sales_orders)order_count = len(sales_orders)# 准备邮件内容mail_body = f"""<h2>上周销售报告 ({date_from} 至 {date_to})</h2><p>订单总数: {order_count}</p><p>销售总额: {total_amount:.2f}</p><p>详细数据请查看附件。</p>"""# 生成Excel报表附件report_data = self._generate_excel_report(sales_orders)attachment_data = {'name': f'销售周报_{date_from}_{date_to}.xlsx','datas': report_data,'res_model': 'sales.report','res_id': 0,'type': 'binary',}attachment_id = self.env['ir.attachment'].create(attachment_data)# 获取销售团队成员sales_team = self.env.ref('sales_team.team_sales_department')recipients = sales_team.member_ids# 发送邮件mail_values = {'subject': f'销售周报 ({date_from} 至 {date_to})','body_html': mail_body,'email_to': ','.join(recipients.mapped('email')),'attachment_ids': [(6, 0, [attachment_id.id])],'auto_delete': True,}self.env['mail.mail'].create(mail_values).send()return Truedef _generate_excel_report(self, sales_orders):"""生成Excel格式的销售报告"""# 这里使用xlsxwriter或其他库生成Excel文件# 简化示例，实际应用中需要完整实现return b'Excel报表内容'  # 返回二进制数据

2. 定时任务配置 (data/ir_cron_data.xml)

<?xml version="1.0" encoding="UTF-8"?>
<odoo><data noupdate="1"><record id="ir_cron_send_weekly_sales_report" model="ir.cron"><!-- 基础配置 --><field name="name">发送销售周报</field><field name="model_id" ref="model_sales_report"/><!-- 执行设置 --><field name="state">code</field><field name="code">model.send_weekly_report()</field><field name="user_id" ref="base.user_root"/><!-- 调度参数 --><field name="interval_number">1</field><field name="interval_type">weeks</field><field name="nextcall" eval="(datetime.now().replace(hour=9, minute=0, second=0) + timedelta(days=(0 - datetime.now().weekday()) % 7)).strftime('%Y-%m-%d %H:%M:%S')"/><field name="numbercall">-1</field><field name="doall">False</field><!-- 优先级 --><field name="priority">5</field><field name="active">True</field></record></data>
</odoo>

3. 模块定义 (manifest.py)

{'name': '销售周报自动发送','version': '1.0','category': 'Sales','summary': '自动发送每周销售报告','description': """每周一早上9点自动向销售团队发送上周的销售统计报告。""",'author': 'Your Company','website': 'https://www.yourcompany.com','depends': ['sale', 'mail'],'data': ['security/ir.model.access.csv','data/ir_cron_data.xml',],'installable': True,'application': False,'auto_install': False,
}

案例二：自动清理过期数据

需求描述

每天凌晨 3 点自动清理 30 天前的系统日志，以防止数据库过大影响性能。

实现步骤

创建一个自定义模块 system_log_cleaner
在模块中定义清理日志的方法
配置 ir.cron 定时任务

代码示例

1. 模型定义 (models/log_cleaner.py)

from odoo import models, api
from datetime import datetime, timedelta
import logging_logger = logging.getLogger(__name__)class LogCleaner(models.Model):_name = 'log.cleaner'_description = '日志清理工具'@api.modeldef clean_old_logs(self):"""清理30天前的系统日志"""# 计算30天前的日期thirty_days_ago = datetime.now() - timedelta(days=30)# 清理邮件日志mail_logs = self.env['mail.mail'].search([('create_date', '<', thirty_days_ago),('state', 'in', ['sent', 'exception', 'cancel'])])mail_count = len(mail_logs)mail_logs.unlink()# 清理HTTP请求日志http_logs = self.env['ir.http.request'].search([('create_date', '<', thirty_days_ago)])http_count = len(http_logs)http_logs.unlink()# 清理审计日志audit_logs = self.env['auditlog.log'].search([('create_date', '<', thirty_days_ago)])audit_count = len(audit_logs)audit_logs.unlink()# 记录清理结果_logger.info('日志清理完成: 删除了 %s 条邮件日志, %s 条HTTP请求日志, %s 条审计日志',mail_count, http_count, audit_count)# 创建清理记录self.env['log.cleaner.history'].create({'date': fields.Date.today(),'mail_logs_count': mail_count,'http_logs_count': http_count,'audit_logs_count': audit_count,'total_count': mail_count + http_count + audit_count,})return Trueclass LogCleanerHistory(models.Model):_name = 'log.cleaner.history'_description = '日志清理历史'date = fields.Date(string='清理日期', required=True)mail_logs_count = fields.Integer(string='邮件日志数量')http_logs_count = fields.Integer(string='HTTP日志数量')audit_logs_count = fields.Integer(string='审计日志数量')total_count = fields.Integer(string='总清理数量')

2. 定时任务配置 (data/ir_cron_data.xml)

<?xml version="1.0" encoding="UTF-8"?>
<odoo><data noupdate="1"><record id="ir_cron_clean_old_logs" model="ir.cron"><!-- 基础配置 --><field name="name">清理过期系统日志</field><field name="model_id" ref="model_log_cleaner"/><!-- 执行设置 --><field name="state">code</field><field name="code">model.clean_old_logs()</field><field name="user_id" ref="base.user_root"/><!-- 调度参数 --><field name="interval_number">1</field><field name="interval_type">days</field><field name="nextcall" eval="(datetime.now().replace(hour=3, minute=0, second=0) + timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')"/><field name="numbercall">-1</field><field name="doall">False</field><!-- 优先级 --><field name="priority">10</field><field name="active">True</field></record></data>
</odoo>

案例三：订单状态自动更新

需求描述

每小时检查一次未支付的销售订单，如果订单创建时间超过 48 小时仍未支付，则自动取消订单并通知客户。

实现步骤

创建一个自定义模块 order_auto_cancel
在模块中定义检查和取消订单的方法
配置 ir.cron 定时任务

代码示例

1. 模型定义 (models/order_auto_cancel.py)

from odoo import models, fields, api
from datetime import datetime, timedeltaclass OrderAutoCancel(models.Model):_name = 'order.auto.cancel'_description = '订单自动取消'@api.modeldef auto_cancel_unpaid_orders(self):"""自动取消48小时未支付的订单"""# 计算48小时前的时间点deadline = datetime.now() - timedelta(hours=48)# 查找需要取消的订单orders_to_cancel = self.env['sale.order'].search([('state', '=', 'draft'),  # 草稿状态('create_date', '<', deadline),('payment_status', '=', 'unpaid')  # 假设有这个字段表示支付状态])# 记录取消的订单数量cancelled_count = len(orders_to_cancel)# 遍历订单并取消for order in orders_to_cancel:# 取消订单order.action_cancel()# 记录取消原因order.message_post(body=f"订单已自动取消：创建时间 {order.create_date} 超过48小时未支付",message_type='comment',subtype_id=self.env.ref('mail.mt_note').id)# 通知客户if order.partner_id.email:template = self.env.ref('order_auto_cancel.email_template_order_auto_cancelled')template.send_mail(order.id, force_send=True)# 记录执行结果self.env['order.cancel.log'].create({'date': fields.Datetime.now(),'orders_count': cancelled_count,'user_id': self.env.user.id,})return Trueclass OrderCancelLog(models.Model):_name = 'order.cancel.log'_description = '订单取消日志'date = fields.Datetime(string='执行时间', required=True)orders_count = fields.Integer(string='取消订单数量')user_id = fields.Many2one('res.users', string='执行用户')

2. 邮件模板 (data/mail_template.xml)

<?xml version="1.0" encoding="UTF-8"?>
<odoo><data noupdate="1"><record id="email_template_order_auto_cancelled" model="mail.template"><field name="name">订单自动取消通知</field><field name="model_id" ref="sale.model_sale_order"/><field name="subject">订单 ${object.name} 已自动取消</field><field name="email_from">${(object.company_id.email or user.email)|safe}</field><field name="email_to">${object.partner_id.email|safe}</field><field name="body_html"><![CDATA[<div style="margin: 0px; padding: 0px;"><p style="margin: 0px; padding: 0px; font-size: 13px;">尊敬的 ${object.partner_id.name}：<br/><br/>您的订单 <strong>${object.name}</strong> 已被系统自动取消，因为该订单在创建后48小时内未完成支付。<br/><br/>订单详情：<ul><li>订单编号: ${object.name}</li><li>创建时间: ${object.create_date}</li><li>订单金额: ${object.amount_total} ${object.currency_id.name}</li></ul><br/>如果您仍希望购买这些产品，请重新下单。<br/><br/>如有任何疑问，请随时与我们联系。<br/><br/>谢谢！</p></div>]]></field><field name="auto_delete" eval="True"/><field name="lang">${object.partner_id.lang}</field></record></data>
</odoo>

3. 定时任务配置 (data/ir_cron_data.xml)

<?xml version="1.0" encoding="UTF-8"?>
<odoo><data noupdate="1"><record id="ir_cron_auto_cancel_unpaid_orders" model="ir.cron"><!-- 基础配置 --><field name="name">自动取消未支付订单</field><field name="model_id" ref="model_order_auto_cancel"/><!-- 执行设置 --><field name="state">code</field><field name="code">model.auto_cancel_unpaid_orders()</field><field name="user_id" ref="base.user_root"/><!-- 调度参数 --><field name="interval_number">1</field><field name="interval_type">hours</field><field name="nextcall" eval="(datetime.now() + timedelta(hours=1)).strftime('%Y-%m-%d %H:%M:%S')"/><field name="numbercall">-1</field><field name="doall">False</field><!-- 优先级 --><field name="priority">5</field><field name="active">True</field></record></data>
</odoo>

案例四：系统健康检查

需求描述

每天早上 6 点进行系统健康检查，包括数据库大小监控、长时间运行的查询检测、系统负载检查等，并将结果发送给系统管理员。

实现步骤

创建一个自定义模块 system_health_check
在模块中定义健康检查的方法
配置 ir.cron 定时任务

代码示例

1. 模型定义 (models/health_check.py)

from odoo import models, fields, api
import psycopg2
import os
import subprocess
import logging_logger = logging.getLogger(__name__)class SystemHealthCheck(models.Model):_name = 'system.health.check'_description = '系统健康检查'@api.modeldef perform_health_check(self):"""执行系统健康检查"""results = {}# 检查数据库大小db_size = self._check_database_size()results['database_size'] = db_size# 检查长时间运行的查询long_queries = self._check_long_running_queries()results['long_queries'] = long_queries# 检查系统负载system_load = self._check_system_load()results['system_load'] = system_load# 检查磁盘空间disk_space = self._check_disk_space()results['disk_space'] = disk_space# 检查活跃用户数active_users = self._check_active_users()results['active_users'] = active_users# 保存检查结果check_record = self.env['system.health.check.log'].create({'date': fields.Datetime.now(),'database_size': db_size['size_mb'],'long_queries_count': len(long_queries),'system_load': system_load['load_avg_1m'],'disk_usage_percent': disk_space['usage_percent'],'active_users_count': active_users['count'],'status': 'normal' if self._is_system_healthy(results) else 'warning',})# 如果有异常情况，发送警告邮件if not self._is_system_healthy(results):self._send_warning_email(results, check_record)return Truedef _check_database_size(self):"""检查数据库大小"""self.env.cr.execute("""SELECT pg_database_size(current_database()) as size""")size_bytes = self.env.cr.fetchone()[0]size_mb = size_bytes / (1024 * 1024)return {'size_bytes': size_bytes,'size_mb': round(size_mb, 2),'size_gb': round(size_mb / 1024, 2)}def _check_long_running_queries(self):"""检查长时间运行的查询（超过30秒）"""# 需要有数据库管理员权限try:# 创建一个新的连接以获取管理员权限db_name = self.env.cr.dbnameconnection = psycopg2.connect(dbname=db_name,user=tools.config.get('db_user', 'odoo'),password=tools.config.get('db_password', 'odoo'),host=tools.config.get('db_host', 'localhost'))cursor = connection.cursor()cursor.execute("""SELECT pid, now() - query_start as duration, queryFROM pg_stat_activityWHERE state = 'active'AND now() - query_start > interval '30 seconds'ORDER BY duration DESC""")long_queries = []for pid, duration, query in cursor.fetchall():long_queries.append({'pid': pid,'duration_seconds': duration.total_seconds(),'query': query})cursor.close()connection.close()return long_queriesexcept Exception as e:_logger.error("检查长时间运行的查询时出错: %s", e)return []def _check_system_load(self):"""检查系统负载"""try:load_avg = os.getloadavg()return {'load_avg_1m': load_avg[0],'load_avg_5m': load_avg[1],'load_avg_15m': load_avg[2]}except Exception as e:_logger.error("检查系统负载时出错: %s", e)return {'load_avg_1m': 0,'load_avg_5m': 0,'load_avg_15m': 0}def _check_disk_space(self):"""检查磁盘空间"""try:# 使用subprocess调用df命令output = subprocess.check_output(['df', '-h', '/']).decode('utf-8')lines = output.strip().split('\n')parts = lines[1].split()# 解析输出total = parts[1]used = parts[2]available = parts[3]usage_percent = int(parts[4].replace('%', ''))return {'total': total,'used': used,'available': available,'usage_percent': usage_percent}except Exception as e:_logger.error("检查磁盘空间时出错: %s", e)return {'total': '0','used': '0','available': '0','usage_percent': 0}def _check_active_users(self):"""检查活跃用户数"""# 查询过去1小时内活跃的用户one_hour_ago = fields.Datetime.now() - timedelta(hours=1)active_users = self.env['res.users'].search([('login_date', '>=', one_hour_ago)])return {'count': len(active_users),'users': active_users.mapped('name')}def _is_system_healthy(self, results):"""根据检查结果判断系统是否健康"""# 设置阈值thresholds = {'database_size_gb': 10,  # 数据库大小超过10GB发出警告'long_queries_count': 3,  # 超过3个长时间运行的查询发出警告'system_load_1m': 5,      # 1分钟负载超过5发出警告'disk_usage_percent': 80  # 磁盘使用率超过80%发出警告}# 检查是否超过阈值warnings = []if results['database_size']['size_gb'] > thresholds['database_size_gb']:warnings.append(f"数据库大小 ({results['database_size']['size_gb']}GB) 超过阈值 ({thresholds['database_size_gb']}GB)")if len(results['long_queries']) > thresholds['long_queries_count']:warnings.append(f"长时间运行的查询数 ({len(results['long_queries'])}) 超过阈值 ({thresholds['long_queries_count']})")if results['system_load']['load_avg_1m'] > thresholds['system_load_1m']:warnings.append(f"系统负载 ({results['system_load']['load_avg_1m']}) 超过阈值 ({thresholds['system_load_1m']})")if results['disk_space']['usage_percent'] > thresholds['disk_usage_percent']:warnings.append(f"磁盘使用率 ({results['disk_space']['usage_percent']}%) 超过阈值 ({thresholds['disk_usage_percent']}%)")return len(warnings) == 0def _send_warning_email(self, results, check_record):"""发送警告邮件给系统管理员"""admin_group = self.env.ref('base.group_system')admin_emails = admin_group.users.mapped('email')if not admin_emails:_logger.warning("没有找到系统管理员的邮箱地址，无法发送警告邮件")return# 准备邮件内容warnings = []thresholds = {'database_size_gb': 10,'long_queries_count': 3,'system_load_1m': 5,'disk_usage_percent': 80}if results['database_size']['size_gb'] > thresholds['database_size_gb']:warnings.append(f"数据库大小 ({results['database_size']['size_gb']}GB) 超过阈值 ({thresholds['database_size_gb']}GB)")if len(results['long_queries']) > thresholds['long_queries_count']:warnings.append(f"长时间运行的查询数 ({len(results['long_queries'])}) 超过阈值 ({thresholds['long_queries_count']})")if results['system_load']['load_avg_1m'] > thresholds['system_load_1m']:warnings.append(f"系统负载 ({results['system_load']['load_avg_1m']}) 超过阈值 ({thresholds['system_load_1m']})")if results['disk_space']['usage_percent'] > thresholds['disk_usage_percent']:warnings.append(f"磁盘使用率 ({results['disk_space']['usage_percent']}%) 超过阈值 ({thresholds['disk_usage_percent']}%)")warning_list = "<ul>" + "".join([f"<li>{w}</li>" for w in warnings]) + "</ul>"mail_body = f"""<h2>系统健康检查警告</h2><p>检查时间: {check_record.date}</p><h3>警告信息:</h3>{warning_list}<h3>详细信息:</h3><p>数据库大小: {results['database_size']['size_gb']}GB</p><p>长时间运行的查询: {len(results['long_queries'])}</p><p>系统负载: {results['system_load']['load_avg_1m']} (1分钟), {results['system_load']['load_avg_5m']} (5分钟), {results['system_load']['load_avg_15m']} (15分钟)</p><p>磁盘使用率: {results['disk_space']['usage_percent']}%</p><p>活跃用户数: {results['active_users']['count']}</p><p>请尽快检查系统状态。</p>"""# 发送邮件mail_values = {'subject': f"[警告] 系统健康检查 - {fields.Date.today()}",'body_html': mail_body,'email_to': ','.join(admin_emails),'auto_delete': True,}self.env['mail.mail'].create(mail_values).send()class SystemHealthCheckLog(models.Model):_name = 'system.health.check.log'_description = '系统健康检查日志'_order = 'date desc'date = fields.Datetime(string='检查时间', required=True)database_size = fields.Float(string='数据库大小(MB)')long_queries_count = fields.Integer(string='长查询数量')system_load = fields.Float(string='系统负载')disk_usage_percent = fields.Integer(string='磁盘使用率(%)')active_users_count = fields.Integer(string='活跃用户数')status = fields.Selection([('normal', '正常'),('warning', '警告')], string='状态', default='normal')notes = fields.Text(string='备注')

2. 定时任务配置 (data/ir_cron_data.xml)

<?xml version="1.0" encoding="UTF-8"?>
<odoo><data noupdate="1"><record id="ir_cron_system_health_check" model="ir.cron"><!-- 基础配置 --><field name="name">系统健康检查</field><field name="model_id" ref="model_system_health_check"/><!-- 执行设置 --><field name="state">code</field><field name="code">model.perform_health_check()</field><field name="user_id" ref="base.user_root"/><!-- 调度参数 --><field name="interval_number">1</field><field name="interval_type">days</field><field name="nextcall" eval="(datetime.now().replace(hour=6, minute=0, second=0) + timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S')"/><field name="numbercall">-1</field><field name="doall">False</field><!-- 优先级 --><field name="priority">1</field><field name="active">True</field></record></data>
</odoo>