302 lines
13 KiB
Python
302 lines
13 KiB
Python
"""课程批量导入向导 —— CSV 模板下载 + 批量创建课程（可选连带章节与章节任务）。
|
||
|
||
模板列（前 8 列为课程本体，后 5 列为可选章节/任务）：
|
||
课程名称、课程编号、所属学院、学期、课程类型、课时、课程大纲、业务情景、
|
||
章节标题、章节说明、章节课时、任务名称、是否完成。
|
||
|
||
- 仅填前 8 列：行为与旧版一致，只创建课程；
|
||
- 填了「章节标题」：自动按 (课程, 章节标题) 找/建章节；
|
||
- 填了「任务名称」：在对应章节下创建任务列表明细行（is_done 由「是否完成」决定）。
|
||
同一课程 / 同一章节多行时自动复用，不重复创建。
|
||
"""
|
||
|
||
import base64
|
||
import csv
|
||
import io
|
||
import logging
|
||
|
||
from odoo import _, api, fields, models
|
||
from odoo.exceptions import UserError
|
||
|
||
_logger = logging.getLogger(__name__)
|
||
|
||
COURSE_TYPE_MAP = {
|
||
'理论': 'theory',
|
||
'实践': 'practice',
|
||
'theory': 'theory',
|
||
'practice': 'practice',
|
||
}
|
||
|
||
COLLEGE_NAME_MAP = {
|
||
'计算机学院': 'computer',
|
||
'软件学院': 'software',
|
||
'商学院': 'business',
|
||
'外国语学院': 'foreign',
|
||
'机械工程学院': 'mechanic',
|
||
'电子信息学院': 'electric',
|
||
'数学学院': 'math',
|
||
'其他': 'other',
|
||
'computer': 'computer',
|
||
'software': 'software',
|
||
'business': 'business',
|
||
'foreign': 'foreign',
|
||
'mechanic': 'mechanic',
|
||
'electric': 'electric',
|
||
'math': 'math',
|
||
'other': 'other',
|
||
}
|
||
|
||
|
||
class CourseImportWizard(models.TransientModel):
|
||
"""课程批量导入向导。"""
|
||
|
||
_name = 'yuthon.school.course.import.wizard'
|
||
_description = '课程批量导入'
|
||
|
||
data_file = fields.Binary(string='上传文件')
|
||
file_name = fields.Char(string='文件名')
|
||
note = fields.Html(
|
||
string='导入说明',
|
||
readonly=True,
|
||
default=lambda self: _(
|
||
'<p><b>模板说明：</b></p>'
|
||
'<ol>'
|
||
'<li>点击下方「下载模板」按钮下载 CSV 模板；</li>'
|
||
'<li>用 Excel / WPS 打开模板，按示例填写数据：<br/>'
|
||
'<b>课程名称</b>、<b>课程编号</b> 为必填；<br/>'
|
||
'<b>所属学院、学期、课程类型、课时、课程介绍、业务情景</b> 为选填；</li>'
|
||
'<li>课程类型可选：理论 / 实践</li>'
|
||
'<li><b>章节 / 任务（可选）</b>：填「章节标题」会自动建/复用章节，'
|
||
'填「任务名称」会在该章节下建一条任务列表明细（是否完成填 是/否）；'
|
||
'同一课程、同一章节多行会自动合并，不会重复创建；</li>'
|
||
'<li>保存为 CSV 后上传并点击「开始导入」；</li>'
|
||
'<li>若课程名称已存在则复用，不会重复创建课程。</li>'
|
||
'</ol>'
|
||
),
|
||
)
|
||
|
||
# ─────────────────── 下载模板 ───────────────────
|
||
|
||
def action_download_template(self):
|
||
"""生成 CSV 模板并提供下载。"""
|
||
header = ['课程名称', '课程编号', '所属学院', '学期',
|
||
'课程类型(理论/实践)', '课时', '课程介绍', '业务情景',
|
||
'章节标题', '章节说明', '章节课时',
|
||
'任务名称', '是否完成']
|
||
sample = ['Python 编程基础', 'PY101', '计算机学院',
|
||
'2025-2026第1学期', '理论', '48',
|
||
'本课程面向零基础学员，以项目驱动方式讲解核心知识',
|
||
'通过电商订单管理场景贯穿课程',
|
||
'第1章 环境搭建', '本章介绍开发环境', '4',
|
||
'安装Python', '否']
|
||
|
||
buf = io.StringIO()
|
||
writer = csv.writer(buf)
|
||
writer.writerow(header)
|
||
writer.writerow(sample)
|
||
|
||
attachment = self.env['ir.attachment'].sudo().create({
|
||
'name': '课程导入模板.csv',
|
||
'datas': base64.b64encode(
|
||
buf.getvalue().encode('utf-8-sig')
|
||
),
|
||
'res_model': self._name,
|
||
'res_id': self.id,
|
||
'type': 'binary',
|
||
'mimetype': 'text/csv',
|
||
})
|
||
return {
|
||
'type': 'ir.actions.act_url',
|
||
'url': f'/web/content/{attachment.id}?download=true',
|
||
'target': 'self',
|
||
}
|
||
|
||
# ─────────────────── 导入 ───────────────────
|
||
|
||
def action_import(self):
|
||
"""解析上传的 CSV 并批量创建课程。"""
|
||
self.ensure_one()
|
||
if not self.data_file:
|
||
raise UserError(_('请先上传文件。'))
|
||
|
||
content = base64.b64decode(self.data_file)
|
||
for encoding in ('utf-8-sig', 'utf-8', 'gbk', 'gb2312'):
|
||
try:
|
||
text = content.decode(encoding)
|
||
break
|
||
except UnicodeDecodeError:
|
||
continue
|
||
else:
|
||
raise UserError(_('无法识别文件编码，请使用 UTF-8 或 GBK 编码的 CSV。'))
|
||
|
||
reader = csv.DictReader(io.StringIO(text))
|
||
if not reader.fieldnames:
|
||
raise UserError(_('CSV 文件为空或格式不正确。'))
|
||
|
||
col_map = self._normalize_columns(reader.fieldnames)
|
||
|
||
created = 0
|
||
reused_course = 0
|
||
created_chapter = 0
|
||
reused_chapter = 0
|
||
created_task = 0
|
||
skipped = 0
|
||
errors = []
|
||
|
||
for row_idx, raw_row in enumerate(reader, start=2):
|
||
row = {col_map.get(k, k): v.strip() if v else '' for k, v in raw_row.items()}
|
||
|
||
name = row.get('name', '').strip()
|
||
code = row.get('code', '').strip()
|
||
if not name:
|
||
skipped += 1
|
||
errors.append(_('第 %d 行：课程名称为必填，跳过。') % row_idx)
|
||
continue
|
||
|
||
try:
|
||
# 课程：按名称优先复用，其次按编号；都不存在才新建
|
||
course = self.env['yuthon.school.course'].sudo().search(
|
||
[('name', '=', name)], limit=1)
|
||
if not course and code:
|
||
course = self.env['yuthon.school.course'].sudo().search(
|
||
[('code', '=', code)], limit=1)
|
||
|
||
if not course:
|
||
vals = {
|
||
'name': name,
|
||
'code': code,
|
||
'state': 'draft',
|
||
}
|
||
college_raw = row.get('college', '')
|
||
if college_raw:
|
||
vals['college'] = COLLEGE_NAME_MAP.get(college_raw, 'other')
|
||
semester = row.get('semester', '')
|
||
if semester:
|
||
vals['semester'] = semester
|
||
course_type_raw = row.get('course_type', '')
|
||
if course_type_raw:
|
||
vals['course_type'] = COURSE_TYPE_MAP.get(
|
||
course_type_raw, 'theory')
|
||
hours_raw = row.get('hours', '')
|
||
if hours_raw:
|
||
try:
|
||
vals['hours'] = int(float(hours_raw))
|
||
except ValueError:
|
||
pass
|
||
introduction = row.get('introduction', '')
|
||
if introduction:
|
||
vals['introduction'] = introduction
|
||
business_scenario = row.get('business_scenario', '')
|
||
if business_scenario:
|
||
vals['business_scenario'] = business_scenario
|
||
course = self.env['yuthon.school.course'].sudo().create(vals)
|
||
created += 1
|
||
else:
|
||
reused_course += 1
|
||
|
||
# 章节：填了章节标题才处理（按 课程+标题 复用）
|
||
chapter_title = row.get('chapter', '').strip()
|
||
chapter = None
|
||
if chapter_title:
|
||
chapter = self.env['yuthon.school.course.chapter'].sudo().search([
|
||
('course_id', '=', course.id),
|
||
('name', '=', chapter_title),
|
||
], limit=1)
|
||
if not chapter:
|
||
chapter_vals = {'course_id': course.id, 'name': chapter_title}
|
||
chapter_desc = row.get('chapter_desc', '').strip()
|
||
if chapter_desc:
|
||
chapter_vals['description'] = chapter_desc
|
||
chapter_hours_raw = row.get('chapter_hours', '').strip()
|
||
if chapter_hours_raw:
|
||
try:
|
||
chapter_vals['class_hours'] = int(float(chapter_hours_raw))
|
||
except ValueError:
|
||
pass
|
||
chapter = self.env['yuthon.school.course.chapter'].sudo().create(
|
||
chapter_vals)
|
||
created_chapter += 1
|
||
else:
|
||
reused_chapter += 1
|
||
|
||
# 任务列表：填了任务名称才在对应章节下创建明细行
|
||
task_name = row.get('task_name', '').strip()
|
||
if task_name and chapter:
|
||
is_done = row.get('is_done', '').strip() in (
|
||
'是', 'y', 'yes', 'true', '1', '完成')
|
||
task_vals = {
|
||
'chapter_id': chapter.id,
|
||
'name': task_name,
|
||
'is_done': is_done,
|
||
'sequence': 10,
|
||
}
|
||
self.env['yuthon.school.course.chapter.task'].sudo().create(
|
||
task_vals)
|
||
created_task += 1
|
||
|
||
except Exception as exc:
|
||
skipped += 1
|
||
errors.append(
|
||
_('第 %d 行「%s」导入失败：%s') % (row_idx, name, str(exc)))
|
||
|
||
summary = _(
|
||
'导入完成！\n'
|
||
'新建课程：%d（复用 %d）\n'
|
||
'新建章节：%d（复用 %d）\n'
|
||
'新建任务：%d\n'
|
||
'跳过：%d'
|
||
) % (created, reused_course, created_chapter, reused_chapter,
|
||
created_task, skipped)
|
||
|
||
if errors:
|
||
summary += '\n' + _('注意事项：\n') + '\n'.join(errors[:10])
|
||
if len(errors) > 10:
|
||
summary += _('\n... 还有 %d 条，详见日志。') % (len(errors) - 10)
|
||
|
||
return {
|
||
'type': 'ir.actions.client',
|
||
'tag': 'display_notification',
|
||
'params': {
|
||
'title': _('导入结果'),
|
||
'message': summary,
|
||
'type': 'info' if not errors else 'warning',
|
||
'sticky': True,
|
||
'next': {'type': 'ir.actions.act_window_close'},
|
||
},
|
||
}
|
||
|
||
def _normalize_columns(self, fieldnames):
|
||
"""列名 → 内部 key 映射。"""
|
||
mapping = {}
|
||
for f in fieldnames:
|
||
f_lower = f.strip().lower()
|
||
if f_lower in ('课程名称', 'name', '课程名'):
|
||
mapping[f] = 'name'
|
||
elif f_lower in ('课程编号', 'code', '课程号', '编号'):
|
||
mapping[f] = 'code'
|
||
elif f_lower in ('所属学院', 'college', '学院'):
|
||
mapping[f] = 'college'
|
||
elif f_lower in ('学期', 'semester'):
|
||
mapping[f] = 'semester'
|
||
elif f_lower in ('课程类型', '课程类型(理论/实践)', 'course_type', '类型'):
|
||
mapping[f] = 'course_type'
|
||
elif f_lower in ('课时', 'hours'):
|
||
mapping[f] = 'hours'
|
||
elif f_lower in ('课程介绍', 'introduction', '课程简介', '课程大纲', 'syllabus', '大纲'):
|
||
mapping[f] = 'introduction'
|
||
elif f_lower in ('业务情景', 'business_scenario', 'scenario'):
|
||
mapping[f] = 'business_scenario'
|
||
elif f_lower in ('章节标题', 'chapter', '章节', 'chapter_name'):
|
||
mapping[f] = 'chapter'
|
||
elif f_lower in ('章节说明', 'chapter_desc', 'chapter_description'):
|
||
mapping[f] = 'chapter_desc'
|
||
elif f_lower in ('章节课时', 'chapter_hours', 'chapter_class_hours'):
|
||
mapping[f] = 'chapter_hours'
|
||
elif f_lower in ('任务名称', 'task_name', '任务', 'task'):
|
||
mapping[f] = 'task_name'
|
||
elif f_lower in ('是否完成', 'is_done', '完成', 'done'):
|
||
mapping[f] = 'is_done'
|
||
else:
|
||
mapping[f] = f # 保留原名
|
||
return mapping
|