@@ -3,6 +3,8 @@ import base64
import csv
import io
import logging
import re
import zipfile
from datetime import datetime , timedelta
from odoo import models , fields , api , _
@@ -10,16 +12,87 @@ from odoo.exceptions import UserError
_logger = logging . getLogger ( __name__ )
# 日志文件解析正则（二进制模式）
# 格式: UserRecordHelper record <user_id> <name> <YYYY-MM-DD HH:MM:SS> <timestamp> Save to database id: <id>
_LOG_RECORD_RE = re . compile (
rb ' UserRecordHelper record \ s+( \ S+) \ s+(.+?) \ s+ '
rb ' ( \ d {4} - \ d {2} - \ d {2} \ s+ \ d {2} : \ d {2} : \ d {2} ) \ s+( \ d+) \ s+Save to database id: \ s*( \ d+) '
)
def _decode_name ( name_bytes ) :
""" 尝试用多种编码解码中文姓名 """
for enc in ( ' utf-8 ' , ' gbk ' , ' gb18030 ' , ' latin-1 ' ) :
try :
return name_bytes . decode ( enc )
except Exception :
continue
return name_bytes . decode ( ' latin-1 ' , errors = ' replace ' )
def _extract_device_sn ( zip_entry_name ) :
""" 从 zip 条目文件名提取设备 SN: _uploads_logs_<SN>_ """
m = re . search ( r ' _uploads_logs_( \ w+)_ ' , zip_entry_name )
return m . group ( 1 ) if m else ' '
def _parse_log_bytes ( raw_bytes , device_sn = ' ' ) :
""" 解析考勤机日志二进制内容，返回记录列表。每条记录: { name, checkin_str, device_sn} """
records = [ ]
for line in raw_bytes . split ( b ' \n ' ) :
m = _LOG_RECORD_RE . search ( line )
if not m :
continue
name_bytes = m . group ( 2 ) . strip ( )
name = _decode_name ( name_bytes )
if not name or name . startswith ( ' F ' ) :
continue
checkin_str = m . group ( 3 ) . decode ( ' ascii ' )
records . append ( {
' name ' : name ,
' checkin_str ' : checkin_str ,
' device_sn ' : device_sn ,
} )
return records
def _parse_csv_bytes ( raw_bytes ) :
""" 解析 CSV 二进制内容，返回记录列表。每条记录: { name, checkin_str, device_sn, remark} """
records = [ ]
try :
content = raw_bytes . decode ( ' utf-8-sig ' )
except Exception :
return records , _ ( " CSV 文件解码失败 " )
reader = csv . DictReader ( io . StringIO ( content ) )
if reader . fieldnames :
reader . fieldnames = [ c . strip ( ) for c in reader . fieldnames ]
required_cols = [ ' employee_id ' , ' checkin_time ' ]
missing = [ c for c in required_cols if c not in ( reader . fieldnames or [ ] ) ]
if missing :
return records , _ ( " 缺少必要列: %(missing)s ，实际列: %(actual)s " ) % {
' missing ' : ' , ' . join ( missing ) ,
' actual ' : ' , ' . join ( reader . fieldnames or [ ] ) ,
}
for row in reader :
name = ( row . get ( ' employee_id ' ) or ' ' ) . strip ( )
checkin_str = ( row . get ( ' checkin_time ' ) or ' ' ) . strip ( )
if not name or not checkin_str :
continue
records . append ( {
' name ' : name ,
' checkin_str ' : checkin_str ,
' device_sn ' : ( row . get ( ' attendance_iot_id ' ) or ' ' ) . strip ( ) ,
' remark ' : ( row . get ( ' remark ' ) or ' ' ) . strip ( ) ,
} )
return records , None
class AttendanceIotImportWizard ( models . TransientModel ) :
_name = " attendance.iot.import.wizard "
_description = " 考勤机日志导入向导 "
log_file = fields . Binary (
string = " CSV 导入 文件" , required = True ,
help = " 上传从日志解析生成的 CSV 文件（包含 employee_id, checkin_time, source, attendance_iot_id, remark 列） "
)
log_filename = fields . Char ( string = " 文件名 " )
zip_file = fields . Binary ( string = " ZIP 压缩包 " , required = True , attachment = False )
zip_filename = fields . Char ( string = " 文件名 " )
note = fields . Text ( string = " 导入备注 " , default = " 日志补录 " )
preview_line_ids = fields . One2many (
' attendance.iot.import.preview.line ' ,
@@ -39,54 +112,69 @@ class AttendanceIotImportWizard(models.TransientModel):
error_message = fields . Text ( string = " 错误信息 " , readonly = True )
def action_parse ( self ) :
""" 解析上传的 CSV 文件 """
""" 解压 zip 并解析其中所有支持的日志 文件 """
self . ensure_one ( )
if not self . log_file :
raise UserError ( _ ( " 请先上传 CSV 文件 ！ " ) )
if not self . zip_file :
raise UserError ( _ ( " 请先上传 ZIP 压缩包 ！ " ) )
# 解码文件内容（支持 utf-8 和 utf-8-sig/BOM）
raw_bytes = base64 . b64decode ( self . zip_file )
# 验证是否是有效 zip
try :
content = base64 . b64decode ( self . log_file ) . decode ( ' utf-8-sig ' )
except Exception as e :
raise UserError ( _ ( " 文件解码失败: %s " ) % str ( e ) )
zf = zipfile . ZipFile ( io . BytesIO ( raw_bytes ) )
except zipfile . BadZipFile :
raise UserError ( _ ( " 上传的文件不是有效的 ZIP 压缩包，请重新压缩后上传！ " ) )
# 解析 CSV
reader = csv . DictReader ( io . StringIO ( content ) )
if reader . fieldnames :
reader . fieldnames = [ c . strip ( ) for c in reader . fieldnames ]
required_cols = [ ' employee_id ' , ' checkin_time ' , ' source ' , ' attendance_iot_id ' ]
missing = [ c for c in required_cols if c not in ( reader . fieldnames or [ ] ) ]
if missing :
raise UserError (
_ ( " CSV 缺少必要列: %s \n 文件实际列名: %s " )
% ( ' , ' . join ( missing ) , ' , ' . join ( reader . fieldnames or [ ] ) )
)
all_records = [ ]
errors = [ ]
file_errors = [ ]
entry_list = [ e for e in zf . infolist ( ) if not e . is_dir ( ) ]
records = [ ]
for row in reader :
name = ( row . get ( ' employee_id ' ) or ' ' ) . strip ( )
checkin_str = ( row . get ( ' checkin_time ' ) or ' ' ) . strip ( )
device_sn = ( row . get ( ' attendance_iot_id ' ) or ' ' ) . strip ( )
remark = ( row . get ( ' remark ' ) or ' ' ) . strip ( )
if not name or not checkin_str :
continue
records . append ( {
' name ' : name ,
' checkin_str ' : checkin_str ,
' device_sn ' : device_sn ,
' remark ' : remark or ( self . note or " 日志补录 " ) ,
} )
if not entry_list :
zf . close ( )
raise UserError ( _ ( " ZIP 压缩包中没有找到任何文件！ " ) )
if not records :
raise UserError ( _ ( " CSV 文件中未找到有效记录！请检查 employee_id 和 checkin_time 列是否有数据。 " ) )
for entry in entry_list :
entry_name = entry . filename
entry_lower = entry_name . lower ( )
entry_bytes = zf . read ( entry )
if entry_lower . endswith ( ' .log ' ) :
device_sn = _extract_device_sn ( entry_name )
parsed = _parse_log_bytes ( entry_bytes , device_sn )
for rec in parsed :
rec [ ' remark ' ] = self . note or " 日志补录 "
all_records . append ( rec )
elif entry_lower . endswith ( ' .csv ' ) :
csv_records , csv_err = _parse_csv_bytes ( entry_bytes )
if csv_err :
file_errors . append ( _ ( " 文件 %s : %s " ) % ( entry_name , csv_err ) )
else :
for rec in csv_records :
if not rec . get ( ' remark ' ) :
rec [ ' remark ' ] = self . note or " 日志补录 "
all_records . append ( rec )
else :
file_errors . append (
_ ( " 跳过不支持的文件: %s （仅支持 .log / .csv） " ) % entry_name
)
zf . close ( )
if not all_records :
msg = _ ( " 所有文件中未找到有效记录！请检查压缩包内文件内容是否正确。 " )
if file_errors :
msg + = " \n " + " \n " . join ( file_errors )
raise UserError ( msg )
# 清空旧预览
self . preview_line_ids . unlink ( )
# 创建预览行
preview_lines = [ ]
errors = [ ]
for rec in records :
for rec in all_records :
employee = self . env [ ' hr.employee ' ] . search ( [ ( ' name ' , ' = ' , rec [ ' name ' ] ) ] , limit = 1 )
if not employee :
errors . append ( _ ( " 未找到员工: %s " ) % rec [ ' name ' ] )
@@ -101,9 +189,9 @@ class AttendanceIotImportWizard(models.TransientModel):
self . write ( {
' preview_line_ids ' : preview_lines ,
' state ' : ' preview ' ,
' total_found ' : len ( records ) ,
' total_found ' : len ( all_records ) ,
' total_skipped ' : len ( errors ) ,
' error_message ' : ' \n ' . join ( errors ) if errors else ' ' ,
' error_message ' : ' \n ' . join ( errors + file_errors ) if ( errors or file_errors ) else ' ' ,
} )
return {
@@ -130,7 +218,6 @@ class AttendanceIotImportWizard(models.TransientModel):
for line in self . preview_line_ids :
try :
# 查找员工
if line . employee_name not in employee_cache :
emp = self . env [ ' hr.employee ' ] . search ( [ ( ' name ' , ' = ' , line . employee_name ) ] , limit = 1 )
employee_cache [ line . employee_name ] = emp
@@ -142,18 +229,15 @@ class AttendanceIotImportWizard(models.TransientModel):
skipped + = 1
continue
# 查找设备
if line . device_sn and line . device_sn not in iot_cache :
iot = self . env [ ' attendance.iot ' ] . search ( [ ( ' name ' , ' = ' , line . device_sn ) ] , limit = 1 )
iot_cache [ line . device_sn ] = iot
else :
iot = iot_cache . get ( line . device_sn )
# 解析时间（CSV 中时间为北京时间 UTC+8）
checkin_naive = datetime . strptime ( line . checkin_str , ' % Y- % m- %d % H: % M: % S ' )
checkin_utc = checkin_naive - timedelta ( hours = 8 )
checkin_native = datetime . strptime ( line . checkin_str , ' % Y- % m- %d % H: % M: % S ' )
checkin_utc = checkin_native - timedelta ( hours = 8 )
# 检查是否已存在（避免重复导入）
existing = self . env [ ' hr.attendance.detail ' ] . search_count ( [
( ' employee_id ' , ' = ' , emp . id ) ,
( ' checkin_time ' , ' = ' , checkin_utc ) ,
@@ -163,7 +247,6 @@ class AttendanceIotImportWizard(models.TransientModel):
skipped + = 1
continue
# 创建记录，跳过企业微信通知
vals = {
' employee_id ' : emp . id ,
' checkin_time ' : checkin_utc ,
李鹏宇