# ============================================================================= # IT智能服务台 — 资产数据服务 # ============================================================================= # 说明：从 Excel 资产清单中读取资产信息，用于审批推送时的设备年限核查。 # - 懒加载 Excel（首次查询时才加载文件） # - 按资产编号查询（遍历所有月度 sheet） # - 计算使用年限（今天 - 开始使用日期） # - 生成审批意见文本 # # Excel 结构： # - 12 个月度 sheet（202501-202512）+ 1 个统计 sheet # - 前 3 行是组织架构汇总行（非资产数据），实际资产数据从第 6 行开始 # - 区分资产行：固定资产编码列（第2列）非 None # # 后续可替换为 API 调用（保持 find_asset / check_device_age 接口不变即可） # ============================================================================= import logging from datetime import date, datetime, timedelta from typing import Optional from openpyxl import load_workbook from app.config import settings logger = logging.getLogger(__name__) class AssetService: """资产数据服务 — 前期从 Excel 读取，后续可替换为 API。 Attributes: excel_path: Excel 资产清单文件路径 """ # 固定资产编码列名 ASSET_CODE_COLUMN = "固定资产编码" # 开始使用日期列名 START_DATE_COLUMN = "开始使用日期" # Excel 列映射（列名 → 列序，从1开始） COLUMN_MAP = { "组织架构": 1, "固定资产编码": 2, "原资产编码": 3, "固定资产名称": 4, "固定资产类别": 5, "实物入账日期": 9, "财务入账日期": 10, "使用状态": 13, "存放地点": 15, "使用部门": 16, "使用人": 17, "交付日期": 40, "开始使用日期": 41, "预计使用年限": 42, } # 月度 sheet 名称前缀（匹配 202501-202512） MONTHLY_SHEET_PREFIX = "2025" # 资产数据起始行（前3行组织架构汇总 + 标题行 + 空行） DATA_START_ROW = 6 def __init__(self, excel_path: str = None): """初始化，懒加载 Excel。 Args: excel_path: Excel 文件路径，默认从 settings.asset_excel_path 读取 """ self.excel_path = excel_path or settings.asset_excel_path self._workbook = None self._loaded = False # ------------------------------------------------------------------ # 内部方法 # ------------------------------------------------------------------ def _ensure_loaded(self): """懒加载 Excel 文件 — 首次调用时才打开文件。""" if self._loaded: return try: self._workbook = load_workbook( self.excel_path, data_only=True, # 读取公式计算后的缓存值 read_only=True, # 只读模式，减少内存占用 ) self._loaded = True logger.info(f"Excel 资产清单加载成功: {self.excel_path}") except Exception as e: logger.error(f"加载 Excel 资产清单失败: {e}") raise def _get_monthly_sheets(self) -> list[str]: """获取所有月度 sheet 名称（202501-202512），按时间升序排列。""" self._ensure_loaded() sheets = [] for name in self._workbook.sheetnames: if name.startswith(self.MONTHLY_SHEET_PREFIX) and len(name) == 6: sheets.append(name) return sorted(sheets) def _build_asset_info(self, row: tuple, sheet_name: str) -> dict: """从 Excel 行数据构建资产信息字典。 Args: row: openpyxl 行数据（values_only=True 的元组） sheet_name: 所属 sheet 名称 Returns: dict: 资产信息字典，key 为列名 """ def safe_get(col_num: int): """安全获取列值（列序从1开始，转为0-based索引）。""" idx = col_num - 1 if idx < len(row): return row[idx] return None info = {"sheet_name": sheet_name} for col_name, col_num in self.COLUMN_MAP.items(): info[col_name] = safe_get(col_num) return info def _parse_date(self, date_value) -> Optional[date]: """解析日期值 — Excel 中的日期可能是 date/datetime/字符串/数字。 Args: date_value: Excel 单元格原始值 Returns: date 对象，解析失败返回 None """ if date_value is None: return None # datetime 对象 → 取 date 部分 if isinstance(date_value, datetime): return date_value.date() # date 对象 → 直接返回 if isinstance(date_value, date): return date_value # 字符串 → 尝试多种格式解析 if isinstance(date_value, str): for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日"): try: return datetime.strptime(date_value.strip(), fmt).date() except ValueError: continue # Excel 数字日期序列号 → 转为 date if isinstance(date_value, (int, float)): try: # Excel 日期序列号：1900-01-01 = 1，修正闰年 bug 后基准为 1899-12-30 return date(1899, 12, 30) + timedelta(days=int(date_value)) except (OverflowError, ValueError): pass logger.warning(f"无法解析日期值: {date_value} (type={type(date_value).__name__})") return None def _format_years(self, years: float) -> str: """将年限（浮点数）格式化为 'X年Y个月' 展示文本。 Args: years: 年限（如 5.17） Returns: str: 如 "5年2个月" """ total_months = int(years * 12) years_part = total_months // 12 months_part = total_months % 12 return f"{years_part}年{months_part}个月" # ------------------------------------------------------------------ # 公开接口 # ------------------------------------------------------------------ def find_asset(self, asset_code: str) -> Optional[dict]: """按资产编号查询资产信息，遍历所有月度 sheet。 从最新月份往前查找，优先返回最新数据。匹配规则：trim + 大小写不敏感。 Args: asset_code: 资产编号 Returns: dict: 资产信息字典，未找到返回 None """ if not asset_code: return None # trim + 大小写不敏感 target_code = asset_code.strip().lower() try: self._ensure_loaded() monthly_sheets = self._get_monthly_sheets() # 从最新月份往前查找（优先返回最新数据） for sheet_name in reversed(monthly_sheets): sheet = self._workbook[sheet_name] # 从第6行开始遍历（前3行组织架构汇总 + 标题/空行） for row in sheet.iter_rows(min_row=self.DATA_START_ROW, values_only=True): if len(row) < 2: continue # 第2列是固定资产编码（列序2 → 索引1） cell_value = row[1] if cell_value is None: continue code_str = str(cell_value).strip().lower() if code_str == target_code: logger.info(f"找到资产: code={asset_code}, sheet={sheet_name}") return self._build_asset_info(row, sheet_name) logger.warning(f"未找到资产编号: {asset_code}") return None except Exception as e: logger.error(f"查询资产信息失败: {e}") return None def check_device_age(self, asset_code: str, threshold_years: int = 5) -> dict: """核查设备使用年限，返回核查结果和审批意见。 流程： 1. 查询资产信息 2. 获取"开始使用日期" 3. 计算使用年限：(今天 - start_date).days / 365.25 4. 判断是否 ≥ 阈值 5. 生成审批意见文本 Args: asset_code: 资产编号 threshold_years: 更换年限阈值，默认 5 年 Returns: dict: 核查结果，包含 found/asset_name/start_date/years_used/ meets_threshold/opinion 等字段 """ # 1. 查询资产信息 asset = self.find_asset(asset_code) if not asset: return { "found": False, "asset_code": asset_code, "opinion": f"未在资产清单中找到编号 {asset_code}，请人工核查。", } # 2. 获取开始使用日期 start_date_raw = asset.get(self.START_DATE_COLUMN) start_date = self._parse_date(start_date_raw) asset_name = asset.get("固定资产名称") or "未知设备" if not start_date: return { "found": True, "asset_code": asset_code, "asset_name": asset_name, "start_date": None, "years_used": None, "years_display": None, "meets_threshold": False, "opinion": f"资产编号 {asset_code}（{asset_name}）无开始使用日期记录，请人工核查。", } # 3. 计算使用年限 today = date.today() days_used = (today - start_date).days years_used = days_used / 365.25 meets = years_used >= threshold_years # 4. 格式化使用年限展示 years_display = self._format_years(years_used) # 5. 生成审批意见 opinion = self._format_opinion( asset_name=asset_name, start_date=start_date, years_used=years_used, threshold=threshold_years, meets=meets, ) return { "found": True, "asset_code": asset_code, "asset_name": asset_name, "start_date": start_date.strftime("%Y-%m-%d"), "years_used": round(years_used, 2), "years_display": years_display, "meets_threshold": meets, "opinion": opinion, "asset_info": asset, } def _format_opinion( self, asset_name: str, start_date: date, years_used: float, threshold: int, meets: bool, ) -> str: """生成审批意见文本。 Args: asset_name: 设备名称 start_date: 开始使用日期 years_used: 使用年限（浮点数） threshold: 更换阈值（年） meets: 是否满足更换条件 Returns: str: 审批意见文本 """ years_display = self._format_years(years_used) if meets: conclusion = f"✅ 已满{threshold}年，符合更换条件" else: conclusion = f"❌ 未满{threshold}年，不符合更换条件" opinion = ( f"设备名称：{asset_name}\n" f"启用日期：{start_date.strftime('%Y-%m-%d')}\n" f"已使用：{years_display}\n" f"核查结论：{conclusion}" ) return opinion def close(self): """关闭 Excel 文件句柄，释放资源。""" if self._workbook: try: self._workbook.close() except Exception: pass self._workbook = None self._loaded = False