347 lines
12 KiB
Python
347 lines
12 KiB
Python
|
# =============================================================================
|
|||
|
|
# 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
|