229 lines
7.9 KiB
Python
229 lines
7.9 KiB
Python
"""课程批量导入向导 —— CSV 模板下载 + 批量创建课程（导入大纲即可）。
|
|
|
|
模板列：课程名称、课程编号、所属学院、学期、课程类型、课时、课程大纲。
|
|
导入时直接创建 yuthon.school.course 记录。
|
|
"""
|
|
|
|
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>保存为 CSV 后上传并点击「开始导入」；</li>'
|
|
'<li>若课程编号已存在则跳过，不会重复创建。</li>'
|
|
'</ol>'
|
|
),
|
|
)
|
|
|
|
# ─────────────────── 下载模板 ───────────────────
|
|
|
|
def action_download_template(self):
|
|
"""生成 CSV 模板并提供下载。"""
|
|
header = ['课程名称', '课程编号', '所属学院', '学期',
|
|
'课程类型(理论/实践)', '课时', '课程大纲']
|
|
sample = ['Python 编程基础', 'PY101', '计算机学院',
|
|
'2025-2026第1学期', '理论', '48',
|
|
'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
|
|
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
|
|
|
|
# 编号去重
|
|
if code:
|
|
existing = self.env['yuthon.school.course'].sudo().search(
|
|
[('code', '=', code)], limit=1)
|
|
if existing:
|
|
skipped += 1
|
|
continue
|
|
|
|
try:
|
|
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'] = float(hours_raw)
|
|
except ValueError:
|
|
pass
|
|
|
|
syllabus = row.get('syllabus', '')
|
|
if syllabus:
|
|
vals['syllabus'] = syllabus
|
|
|
|
self.env['yuthon.school.course'].sudo().create(vals)
|
|
created += 1
|
|
|
|
except Exception as exc:
|
|
skipped += 1
|
|
errors.append(
|
|
_('第 %d 行「%s」导入失败：%s') % (row_idx, name, str(exc)))
|
|
|
|
summary = _(
|
|
'导入完成！\n'
|
|
'新建课程：%d\n'
|
|
'跳过：%d'
|
|
) % (created, 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 ('课程大纲', 'syllabus', '大纲'):
|
|
mapping[f] = 'syllabus'
|
|
else:
|
|
mapping[f] = f # 保留原名
|
|
return mapping
|