feat: 2026-07-11 全量更新 - 代办集成+会议室预定+知识迭代修复+UI统一+Bug修复
== 已部署上线 (9项) == - 代办事项真实数据源集成 (企微审批API 8bug修复链) - H5/坐席端 Logo样式统一+绿色背景 - 视频引导页修复 (localStorage key v2) - 坐席端 v9 Vue版本修复 (ElMessage._context) - 截图按钮 v10 修复 (getDisplayMedia user gesture) - 扫码样式恢复+H5扫码登录跳转修复 - H5截图快捷键提示 == 代码完成待部署 (3项) == - 知识迭代3Bug修复 (#8 POST端点/#7 MERGE幂等/#6 过期检查) - 会议室预定-小鱼易联终端 (40文件, 40/40测试通过) - IT资产升级审批推送 (asset_service.py) == 需求文档 (2项) == - 坐席端AI辅助消息框-PRD (4项新功能确认) - 坐席端布局优化建议 v2.0 (7天计划) == 新增文档 == - 日报-2026-07-11.md - 知识迭代Bug修复报告-20260711.md - 会议室预定-部署指南.md - CHANGELOG.md 更新 == 测试 == - test_todo_integration.py: 40/40 - test_meetingroom.py: 40/40 - test_bugfix_ki_suggestions.py: 21/21
This commit is contained in:
@@ -0,0 +1,346 @@
|
||||
# =============================================================================
|
||||
# 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
|
||||
Reference in New Issue
Block a user