# encoding: UTF-8 from sqlalchemy import func from ..model.caseModel import CaseReview, CaseSnapshot, Module, TestCase from logger import logger class CaseDao(object): """用例域通用 DAO，复用模块、用例、快照、评审的基础操作。""" @staticmethod def create(session, model_cls, add_info): """创建记录并提交事务。""" obj = model_cls(**add_info) session.add(obj) err = session.done(close=False) if err: logger.warning(f'{model_cls.__name__}新增失败！{err}') return 0, f'新增失败！{err}' return obj.id, '' @staticmethod def update_by_id(session, model_cls, obj_id, update_info, soft_delete=True): filters = [model_cls.id == int(obj_id)] if soft_delete and hasattr(model_cls, 'is_delete'): filters.append(model_cls.is_delete == 0) update_res = session.query(model_cls).filter(*filters).update(update_info) err = session.done(close=False) if err: logger.error(f'{model_cls.__name__}更新失败！id: {obj_id}, err: {err}') return 0, f'更新失败！{err}' if not update_res: return 0, '未查询到对应记录！' return int(obj_id), '' @staticmethod def get_by_id(session, model_cls, obj_id, soft_delete=True): filters = [model_cls.id == int(obj_id)] if soft_delete and hasattr(model_cls, 'is_delete'): filters.append(model_cls.is_delete == 0) return session.query(model_cls).filter(*filters).first() @staticmethod def list_by_filters(session, model_cls, filter_list, page=1, limit=20, order_column=None): query = session.query(model_cls).filter(*filter_list) if hasattr(model_cls, 'is_delete'): query = query.filter(model_cls.is_delete == 0) total = query.count() if order_column is not None: query = query.order_by(order_column.desc()) rets = query.offset((int(page) - 1) * int(limit)).limit(int(limit)).all() return rets, total @staticmethod def delete_by_id(session, model_cls, obj_id): return CaseDao.update_by_id(session, model_cls, obj_id, {'is_delete': 1}) @staticmethod def next_case_key(session, project_id, module_id=None, product_id=None): from ..model.productModel import Product from ..model.projectModel import Project from ..model.caseModel import Module product_abbr = '' if product_id: product = session.query(Product).filter(Product.id == int(product_id), Product.is_delete == 0).first() if product and product.name: product_abbr = CaseDao._generate_abbreviation(product.name) project_abbr = '' project = session.query(Project).filter(Project.id == int(project_id), Project.is_delete == 0).first() if project and project.name: project_abbr = CaseDao._generate_abbreviation(project.name) module_abbr = '' if module_id: module = session.query(Module).filter(Module.id == int(module_id), Module.is_delete == 0).first() if module and module.name: module_abbr = CaseDao._generate_abbreviation(module.name) parts = ['TC'] if product_abbr: parts.append(product_abbr) if project_abbr: parts.append(project_abbr) if module_abbr: parts.append(module_abbr) prefix = '-'.join(parts) count_num = session.query(func.count(TestCase.id)).filter( TestCase.project_id == int(project_id), TestCase.is_delete == 0, TestCase.case_key.like(f'{prefix}-%') ).scalar() or 0 return '{}-{:03d}'.format(prefix, count_num + 1) @staticmethod def _generate_abbreviation(name): import re chinese_pattern = re.compile(r'[\u4e00-\u9fff]+') english_pattern = re.compile(r'[a-zA-Z]+') chinese_chars = chinese_pattern.findall(name) if chinese_chars: full_chinese = ''.join(chinese_chars) abbr = ''.join([CaseDao._get_pinyin_first_char(c) for c in full_chinese[:4]]) abbr = abbr.lower() if len(abbr) < 2: abbr = abbr.ljust(2, 'n') return abbr english_words = english_pattern.findall(name) if english_words: abbr = english_words[0].lower()[:4] if len(abbr) < 2: abbr = abbr.ljust(2, 'x') return abbr abbr = name.lower()[:4] if len(abbr) < 2: abbr = abbr.ljust(2, 'x') return abbr @staticmethod def _get_pinyin_first_char(char): pinyin_map = { '智': 'Z', '慧': 'H', '运': 'Y', '营': 'Y', '报': 'B', '关': 'G', '工': 'G', '作': 'Z', '台': 'T', '测': 'C', '试': 'S', '用': 'Y', '例': 'L', '产': 'C', '品': 'P', '项': 'X', '目': 'M', '模': 'M', '块': 'K', '管': 'G', '理': 'L', '系': 'X', '统': 'T', '功': 'G', '能': 'N', '页': 'Y', '面': 'M', '查': 'C', '询': 'X', '添': 'T', '加': 'J', '编': 'B', '辑': 'J', '删': 'S', '除': 'C', '导': 'D', '入': 'R', '导': 'D', '出': 'C', '批': 'P', '量': 'L', '设': 'S', '置': 'Z', '配': 'P', '置': 'Z', '权': 'Q', '限': 'X', '角': 'J', '色': 'S', '用': 'Y', '户': 'H', '组': 'Z', '织': 'Z', '计': 'J', '划': 'H', '执': 'Z', '行': 'X', '报': 'B', '告': 'G', '统': 'T', '计': 'J', '首': 'S', '页': 'Y', '仪': 'Y', '表': 'B', '烟': 'Y', '冒': 'M', '回': 'H', '归': 'G', '集': 'J', '成': 'C', '接': 'J', '口': 'K', '安': 'A', '全': 'Q', '日': 'R', '志': 'Z', '监': 'J', '控': 'K', '优': 'Y', '化': 'H', '性': 'X', '能': 'N', '首': 'S', '尾': 'W', '开': 'K', '发': 'F', '测': 'C', '运': 'Y', '维': 'W', '设': 'S', '计': 'J', '研': 'Y', '发': 'F', '部': 'B', '组': 'Z', '个': 'G', '人': 'R', '公': 'G', '司': 'S', '有': 'Y', '限': 'X', '责': 'Z', '任': 'R', '股': 'G', '份': 'F', '集': 'J', '团': 'T', '科': 'K', '技': 'J', '网': 'W', '络': 'L', '信': 'X', '息': 'X', '软': 'R', '件': 'J', '系': 'X', '统': 'T', '解': 'J', '决': 'J', '方': 'F', '案': 'A', '服': 'F', '务': 'W', '支': 'Z', '持': 'C', '培': 'P', '训': 'X', '咨': 'Z', '询': 'X', '销': 'X', '售': 'S', '市': 'S', '场': 'C', '运': 'Y', '营': 'Y', '管': 'G', '理': 'L', '财': 'C', '务': 'W', '人': 'R', '力': 'L', '资': 'Z', '源': 'Y', '行': 'X', '政': 'Z', '法': 'F', '律': 'L', '合': 'H', '规': 'G', '质': 'Z', '量': 'L', '安': 'A', '全': 'Q', '环': 'H', '境': 'J', '职': 'Z', '能': 'N', '模': 'M', '块': 'K', '页': 'Y', '面': 'M', '窗': 'C', '口': 'K', '表': 'B', '单': 'D', '按': 'A', '钮': 'N', '链': 'L', '接': 'J', '图': 'T', '标': 'B', '菜': 'C', '单': 'D', '导': 'D', '航': 'H', '搜': 'S', '索': 'S', '过': 'G', '滤': 'L', '排': 'P', '序': 'X', '分': 'F', '页': 'Y', '导': 'D', '入': 'R', '导': 'D', '出': 'C', '打': 'D', '印': 'Y', '导': 'D', '出': 'C', '删': 'S', '除': 'C', '复': 'F', '制': 'Z', '粘': 'N', '贴': 'T', '剪': 'J', '切': 'Q', '保': 'B', '存': 'C', '取': 'Q', '消': 'X', '确': 'Q', '认': 'R', '提': 'T', '交': 'J', '审': 'S', '核': 'H', '批': 'P', '准': 'Z', '拒': 'J', '绝': 'J', '返': 'F', '回': 'H', '修': 'X', '改': 'G', '查': 'C', '看': 'K', '详': 'X', '情': 'Q', '列': 'L', '表': 'B', '统': 'T', '计': 'J', '图': 'T', '表': 'B', '报': 'B', '告': 'G', '日': 'R', '志': 'Z', '备': 'B', '注': 'Z', '描': 'M', '述': 'S', '名': 'M', '称': 'C', '编': 'B', '号': 'H', '类': 'L', '型': 'X', '状': 'Z', '态': 'T', '优': 'Y', '先': 'X', '级': 'J', '标': 'B', '签': 'Q', '关': 'G', '键': 'J', '词': 'C', '描': 'M', '述': 'S', '附': 'F', '件': 'J', '链': 'L', '接': 'J', '时': 'S', '间': 'J', '日': 'R', '期': 'Q', '数': 'S', '量': 'L', '金': 'J', '额': 'E', '价': 'J', '格': 'G', '数': 'S', '据': 'J', '库': 'K', '服': 'F', '务': 'W', '器': 'Q', '线': 'X', '程': 'C', '进': 'J', '度': 'D', '速': 'S', '度': 'D', '效': 'X', '率': 'L', '性': 'X', '能': 'N', '稳': 'W', '定': 'D', '可': 'K', '靠': 'K', '安': 'A', '全': 'Q', '兼': 'J', '容': 'R', '扩': 'K', '展': 'Z', '升': 'S', '级': 'J', '更': 'G', '新': 'X', '修': 'X', '补': 'B', '修': 'X', '复': 'F', '优': 'Y', '化': 'H', '重': 'Z', '构': 'G', '改': 'G', '造': 'Z', '移': 'Y', '植': 'Z', '集': 'J', '成': 'C', '接': 'J', '口': 'K', '调': 'T', '试': 'S', '测': 'C', '试': 'S', '验': 'Y', '证': 'Z', '确': 'Q', '认': 'R', '回': 'H', '归': 'G', '演': 'Y', '示': 'S', '培': 'P', '训': 'X', '指': 'Z', '导': 'D', '支': 'Z', '持': 'C', '帮': 'B', '助': 'Z', '反': 'F', '馈': 'K', '投': 'T', '诉': 'S', '建': 'J', '议': 'Y', '评': 'P', '独': 'D', '立': 'L', '站': 'Z', '采': 'C', '购': 'G', '供': 'G', '应': 'Y', '商': 'S', '价': 'J', '满': 'M', '意': 'Y', '度': 'D' } return pinyin_map.get(char, 'N') @staticmethod def next_snapshot_version(session, case_id): """生成用例快照版本号。""" max_version = session.query(func.max(CaseSnapshot.version)).filter(CaseSnapshot.case_id == int(case_id)).scalar() or 0 return int(max_version) + 1 @staticmethod def module_model(): return Module @staticmethod def case_model(): return TestCase @staticmethod def snapshot_model(): return CaseSnapshot @staticmethod def review_model(): return CaseReview @staticmethod def get_module_name_map(session, module_ids): if not module_ids: return {} module_items = session.query(Module).filter(Module.id.in_(module_ids), Module.is_delete == 0).all() return {module.id: module.name for module in module_items}