Files
QueryRewrite/rag2_0/api/create_qingdan_dinge_database.py
T

592 lines
27 KiB
Python

import os
import sys
import sqlite3
import pandas as pd
from openpyxl import load_workbook
import logging
import numpy as np
sys.path.append(os.getcwd())
from rag2_0.tool.ModelTool import XinferenceEmbeddings
from langchain_community.vectorstores import SQLiteVSS
class ExcelToSQLiteProcessor:
"""Excel文件到SQLite数据库的处理器"""
# 定额库表名映射
ding_e_table_names = {
"定额资源库属性": "ding_e_zyk_shuxing",
"定额目录": "ding_e_mulu",
"定额子目": "ding_e_zimu"
}
# 清单库表名映射
qing_dan_table_names = {
"资源库属性": "qd_zyk_shuxing",
"清单目录": "qd_mulu",
"清单子目": "qd_zimu"
}
# 定额库字段映射
ding_e_field_map = {
"资源库名称": "zyk_mc",
"发布时间": "fb_sj",
"适用范围": "sy_fw",
"章节码": "zj_m",
"父章节(章节码)": "fzj_m",
"名称": "mc",
"编码": "bm",
"单位": "dw",
"基价不含税": "jj_bhs",
"基价含税": "jj_hs",
"人工费基价不含税": "rgf_jj_bhs",
"材料费基价不含税": "clf_jj_bhs",
"机械费基价不含税": "jxf_jj_bhs",
"人工费基价含税": "rgf_jj_hs",
"材料费基价含税": "clf_jj_hs",
"机械费基价含税": "jxf_jj_hs",
"人工工日": "rg_gr",
"定额类型": "de_lx",
"工作内容": "gz_nr"
}
# 清单库字段映射
qing_dan_field_map = {
"资源库名称": "zyk_mc",
"发布时间": "fb_sj",
"适用范围": "sy_fw",
"章节码": "zj_m",
"父章节": "fzj_m",
"名称": "mc",
"编码": "bm",
"单位": "dw",
"工作内容": "gz_nr",
"计算规则": "js_gz",
"项目特征": "xm_tz",
"特征值": "tz_z"
}
def __init__(self, db_path):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self.cursor = self.conn.cursor()
self._create_tables()
def _safe_str_convert(self, value):
"""安全地将值转换为字符串"""
if value is None or pd.isna(value):
return ""
return str(value).strip()
def _create_tables(self):
"""创建数据库表结构 - 所有字段都使用TEXT类型"""
print("正在创建数据库表结构...")
# 创建定额库表 - 所有字段都改为TEXT类型
self.cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {self.ding_e_table_names["定额资源库属性"]} (
{self.ding_e_field_map["资源库名称"]} TEXT,
{self.ding_e_field_map["发布时间"]} TEXT,
{self.ding_e_field_map["适用范围"]} TEXT
)
""")
self.cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {self.ding_e_table_names["定额目录"]} (
{self.ding_e_field_map["章节码"]} TEXT,
{self.ding_e_field_map["父章节(章节码)"]} TEXT,
{self.ding_e_field_map["名称"]} TEXT,
{self.ding_e_field_map["资源库名称"]} TEXT,
PRIMARY KEY ({self.ding_e_field_map["资源库名称"]}, {self.ding_e_field_map["章节码"]}, {self.ding_e_field_map["名称"]})
)
""")
self.cursor.execute(f"""
CREATE TABLE IF NOT EXISTS {self.ding_e_table_names["定额子目"]} (
{self.ding_e_field_map["章节码"]} TEXT,
{self.ding_e_field_map["编码"]} TEXT,
{self.ding_e_field_map["名称"]} TEXT,
{self.ding_e_field_map["单位"]} TEXT,
{self.ding_e_field_map["基价不含税"]} TEXT,
{self.ding_e_field_map["基价含税"]} TEXT,
{self.ding_e_field_map["人工费基价不含税"]} TEXT,
{self.ding_e_field_map["材料费基价不含税"]} TEXT,
{self.ding_e_field_map["机械费基价不含税"]} TEXT,
{self.ding_e_field_map["人工费基价含税"]} TEXT,
{self.ding_e_field_map["材料费基价含税"]} TEXT,
{self.ding_e_field_map["机械费基价含税"]} TEXT,
{self.ding_e_field_map["人工工日"]} TEXT,
{self.ding_e_field_map["定额类型"]} TEXT,
{self.ding_e_field_map["工作内容"]} TEXT,
{self.ding_e_field_map["资源库名称"]} TEXT,
PRIMARY KEY ({self.ding_e_field_map["资源库名称"]}, {self.ding_e_field_map["章节码"]}, {self.ding_e_field_map["编码"]}, {self.ding_e_field_map["名称"]})
)
""")
# 创建清单库表 - 所有字段都改为TEXT类型
self.cursor.execute(f'''
CREATE TABLE IF NOT EXISTS {self.qing_dan_table_names["资源库属性"]} (
{self.qing_dan_field_map["资源库名称"]} TEXT PRIMARY KEY,
{self.qing_dan_field_map["发布时间"]} TEXT,
{self.qing_dan_field_map["适用范围"]} TEXT
)
''')
self.cursor.execute(f'''
CREATE TABLE IF NOT EXISTS {self.qing_dan_table_names["清单目录"]} (
{self.qing_dan_field_map["资源库名称"]} TEXT,
{self.qing_dan_field_map["章节码"]} TEXT,
{self.qing_dan_field_map["父章节"]} TEXT,
{self.qing_dan_field_map["名称"]} TEXT,
PRIMARY KEY ({self.qing_dan_field_map["资源库名称"]}, {self.qing_dan_field_map["章节码"]}, {self.qing_dan_field_map["名称"]})
)
''')
self.cursor.execute(f'''
CREATE TABLE IF NOT EXISTS {self.qing_dan_table_names["清单子目"]} (
{self.qing_dan_field_map["资源库名称"]} TEXT,
{self.qing_dan_field_map["章节码"]} TEXT,
{self.qing_dan_field_map["编码"]} TEXT,
{self.qing_dan_field_map["名称"]} TEXT,
{self.qing_dan_field_map["单位"]} TEXT,
{self.qing_dan_field_map["工作内容"]} TEXT,
{self.qing_dan_field_map["计算规则"]} TEXT,
{self.qing_dan_field_map["项目特征"]} TEXT,
{self.qing_dan_field_map["特征值"]} TEXT,
PRIMARY KEY ({self.qing_dan_field_map["资源库名称"]}, {self.qing_dan_field_map["章节码"]}, {self.qing_dan_field_map["编码"]}, {self.qing_dan_field_map["名称"]})
)
''')
print("数据库表结构创建完成")
def process_ding_e_files(self, ding_e_base_dir):
"""处理定额库Excel文件"""
print("=" * 50)
print("开始处理定额库文件...")
print("=" * 50)
if not os.path.exists(ding_e_base_dir):
print(f"定额库目录不存在: {ding_e_base_dir}")
return
# 遍历 Excel 文件
for file_name in os.listdir(ding_e_base_dir):
if not file_name.lower().endswith((".xls", ".xlsx")):
continue
file_path = os.path.join(ding_e_base_dir, file_name)
print(f"正在处理定额库文件: {file_path}")
try:
df_attr = pd.read_excel(file_path, sheet_name="资源库属性", dtype=str)
df_mulu = pd.read_excel(file_path, sheet_name="定额目录", dtype=str)
df_zimu = pd.read_excel(file_path, sheet_name="定额子目", dtype=str)
except Exception as e:
print(f"读取 {file_name} 出错: {e}")
continue
# 提取资源库属性
attr_dict = pd.Series(df_attr["属性值"].values, index=df_attr["资源库属性"]).to_dict()
zyk_name = self._safe_str_convert(attr_dict.get("资源库名称", ""))
pub_time = self._safe_str_convert(attr_dict.get("发布时间", ""))
scope = self._safe_str_convert(attr_dict.get("适用范围", ""))
self.cursor.execute(
f"INSERT INTO {self.ding_e_table_names['定额资源库属性']} VALUES (?, ?, ?)",
(zyk_name, pub_time, scope)
)
# 定额目录 - 转换所有数据为字符串
df_mulu_copy = df_mulu.copy()
df_mulu_copy.rename(columns=self.ding_e_field_map, inplace=True)
df_mulu_copy[self.ding_e_field_map["资源库名称"]] = zyk_name
# 将所有列转换为字符串
for col in df_mulu_copy.columns:
df_mulu_copy[col] = df_mulu_copy[col].apply(self._safe_str_convert)
df_mulu_copy.to_sql(self.ding_e_table_names["定额目录"], self.conn, if_exists="append", index=False)
# 定额子目 - 转换所有数据为字符串
df_zimu_copy = df_zimu.copy()
df_zimu_copy.rename(columns=self.ding_e_field_map, inplace=True)
df_zimu_copy[self.ding_e_field_map["资源库名称"]] = zyk_name
# 将所有列转换为字符串
for col in df_zimu_copy.columns:
df_zimu_copy[col] = df_zimu_copy[col].apply(self._safe_str_convert)
df_zimu_copy.to_sql(self.ding_e_table_names["定额子目"], self.conn, if_exists="append", index=False)
print(f" 成功处理定额库文件: {file_name}")
print("定额库文件处理完成")
def parse_merged_excel_sheet(self, file_path, sheet_name):
"""
解析包含合并单元格的Excel表格
支持工作内容、项目特征、特征值既可能是合并单元格也可能是非合并单元格的情况
"""
# 使用openpyxl读取工作簿以获取合并单元格信息
wb = load_workbook(file_path, data_only=True)
ws = wb[sheet_name]
# 获取所有合并单元格的范围
merged_ranges = list(ws.merged_cells.ranges)
# 创建一个字典来存储合并单元格的值
merged_cells_dict = {}
# 为每个合并单元格范围创建映射
for merged_range in merged_ranges:
# 获取合并单元格左上角的值
top_left_cell = ws[merged_range.coord.split(':')[0]]
value = top_left_cell.value
# 将这个值应用到合并范围内的所有单元格
for row in range(merged_range.min_row, merged_range.max_row + 1):
for col in range(merged_range.min_col, merged_range.max_col + 1):
merged_cells_dict[(row, col)] = value
# 将数据转换为二维数组进行处理
data_array = []
# 遍历所有行和列,应用合并单元格的值
for row_idx in range(1, ws.max_row + 1): # 从第1行开始(包含表头)
row_data = []
for col_idx in range(1, min(ws.max_column + 1, 9)): # 只取前8列
cell_key = (row_idx, col_idx)
if cell_key in merged_cells_dict:
cell_value = merged_cells_dict[cell_key]
else:
cell = ws.cell(row=row_idx, column=col_idx)
cell_value = cell.value
row_data.append(cell_value)
data_array.append(row_data)
# 跳过表头行
if len(data_array) > 1:
data_array = data_array[1:]
# 处理数据:基于章节码、编码、名称进行分组
processed_data = []
# 存储所有数据行,用于后续分组处理
all_rows = []
for row_data in data_array:
if len(row_data) < 8:
continue
# 将所有数据转换为字符串
章节码, 编码, 名称, 单位, 工作内容, 计算规则, 项目特征, 特征值 = [
self._safe_str_convert(cell) for cell in row_data[:8]
]
all_rows.append({
'章节码': 章节码,
'编码': 编码,
'名称': 名称,
'单位': 单位,
'工作内容': 工作内容,
'计算规则': 计算规则,
'项目特征': 项目特征,
'特征值': 特征值
})
# 基于章节码、编码、名称进行分组处理
grouped_data = {}
for row in all_rows:
# 创建分组键
group_key = (row['章节码'], row['编码'], row['名称'])
# 如果章节码、编码、名称都有值,则作为主记录
if row['章节码'] and row['编码'] and row['名称']:
if group_key not in grouped_data:
grouped_data[group_key] = {
'章节码': row['章节码'],
'编码': row['编码'],
'名称': row['名称'],
'单位': row['单位'],
'工作内容': [],
'计算规则': row['计算规则'],
'项目特征': [],
'特征值': []
}
# 更新单位和计算规则(如果当前行有值且之前没有值)
if row['单位'] and not grouped_data[group_key]['单位']:
grouped_data[group_key]['单位'] = row['单位']
if row['计算规则'] and not grouped_data[group_key]['计算规则']:
grouped_data[group_key]['计算规则'] = row['计算规则']
# 查找该行属于哪个组(找最近的有效组)
target_group = None
if row['章节码'] and row['编码'] and row['名称']:
target_group = group_key
else:
# 如果当前行没有完整的分组信息,查找最近的有效组
# 这里采用向上查找的策略,找到最近的有效分组
for existing_key in reversed(list(grouped_data.keys())):
# 如果章节码匹配(或为空),则认为属于该组
if (not row['章节码'] or row['章节码'] == existing_key[0] or
not row['编码'] or row['编码'] == existing_key[1] or
not row['名称'] or row['名称'] == existing_key[2]):
target_group = existing_key
break
# 如果还找不到,使用最后一个组
if not target_group and grouped_data:
target_group = list(grouped_data.keys())[-1]
# 将工作内容、项目特征、特征值添加到对应的组
if target_group and target_group in grouped_data:
if row['工作内容']:
# 避免重复添加
if row['工作内容'] not in grouped_data[target_group]['工作内容']:
grouped_data[target_group]['工作内容'].append(row['工作内容'])
if row['项目特征']:
if row['项目特征'] not in grouped_data[target_group]['项目特征']:
grouped_data[target_group]['项目特征'].append(row['项目特征'])
if row['特征值']:
if row['特征值'] not in grouped_data[target_group]['特征值']:
grouped_data[target_group]['特征值'].append(row['特征值'])
# 将分组后的数据转换为最终格式
for group_key, group_data in grouped_data.items():
processed_data.append({
'章节码': group_data['章节码'],
'编码': group_data['编码'],
'名称': group_data['名称'],
'单位': group_data['单位'],
'工作内容': '\n'.join(group_data['工作内容']),
'计算规则': group_data['计算规则'],
'项目特征': '\n'.join(group_data['项目特征']),
'特征值': '\n'.join(group_data['特征值'])
})
return processed_data
def process_qing_dan_files(self, qing_dan_base_dir):
"""处理清单库Excel文件"""
print("=" * 50)
print("开始处理清单库文件...")
print("=" * 50)
if not os.path.exists(qing_dan_base_dir):
print(f"清单库目录不存在: {qing_dan_base_dir}")
return
try:
# 获取目录下的所有Excel文件
excel_files = [f for f in os.listdir(qing_dan_base_dir) if f.endswith('.xlsx') or f.endswith('.xls')]
for excel_file in excel_files:
file_path = os.path.join(qing_dan_base_dir, excel_file)
print(f"处理清单库文件: {excel_file}")
# 使用openpyxl加载工作簿以检查sheet名称
wb = load_workbook(file_path, read_only=True, data_only=True)
sheet_names = wb.sheetnames
# 检查是否包含所需的三个页签
required_sheets = ['资源库属性', '清单目录', '清单子目']
if not all(sheet in sheet_names for sheet in required_sheets):
print(f"警告: {excel_file} 不包含所需的全部页签,跳过此文件")
continue
# 处理资源库属性页签
try:
prop_df = pd.read_excel(file_path, sheet_name='资源库属性', header=None, dtype=str)
# 找到属性和值的列
prop_df.columns = ['属性', '值'] if len(prop_df.columns) >= 2 else ['属性'] + [f'值{i}' for i in range(len(prop_df.columns)-1)]
# 提取资源库名称、发布时间和适用范围 - 转换为字符串
资源库名称 = self._safe_str_convert(prop_df.loc[prop_df['属性'] == '资源库名称', '值'].iloc[0] if '资源库名称' in prop_df['属性'].values else excel_file.split('.')[0])
发布时间 = self._safe_str_convert(prop_df.loc[prop_df['属性'] == '发布时间', '值'].iloc[0] if '发布时间' in prop_df['属性'].values else '')
适用范围 = self._safe_str_convert(prop_df.loc[prop_df['属性'] == '适用范围', '值'].iloc[0] if '适用范围' in prop_df['属性'].values else '')
# 插入资源库属性
self.cursor.execute(
f"INSERT OR REPLACE INTO {self.qing_dan_table_names['资源库属性']} ({self.qing_dan_field_map['资源库名称']}, {self.qing_dan_field_map['发布时间']}, {self.qing_dan_field_map['适用范围']}) VALUES (?, ?, ?)",
(资源库名称, 发布时间, 适用范围)
)
# 处理清单目录页签
目录_df = pd.read_excel(file_path, sheet_name='清单目录', dtype=str)
for _, row in 目录_df.iterrows():
if pd.notna(row['章节码']): # 确保章节码不为空
# 将所有数据转换为字符串
章节码_str = self._safe_str_convert(row['章节码'])
父章节_str = self._safe_str_convert(row['父章节(章节码)']) if pd.notna(row['父章节(章节码)']) else ''
名称_str = self._safe_str_convert(row['名称']) if pd.notna(row['名称']) else ''
self.cursor.execute(
f"INSERT OR REPLACE INTO {self.qing_dan_table_names['清单目录']} ({self.qing_dan_field_map['资源库名称']}, {self.qing_dan_field_map['章节码']}, {self.qing_dan_field_map['父章节']}, {self.qing_dan_field_map['名称']}) VALUES (?, ?, ?, ?)",
(资源库名称, 章节码_str, 父章节_str, 名称_str)
)
# 处理清单子目页签 - 使用改进的合并单元格处理函数
print(f" 正在处理清单子目页签...")
processed_data = self.parse_merged_excel_sheet(file_path, '清单子目')
# 将处理后的数据插入数据库
for data in processed_data:
if data['章节码'] and data['编码'] and data['名称']: # 确保主要字段不为空
# 所有数据都已经在parse_merged_excel_sheet中转换为字符串
self.cursor.execute(
f"INSERT OR REPLACE INTO {self.qing_dan_table_names['清单子目']} ({self.qing_dan_field_map['资源库名称']}, {self.qing_dan_field_map['章节码']}, {self.qing_dan_field_map['编码']}, {self.qing_dan_field_map['名称']}, {self.qing_dan_field_map['单位']}, {self.qing_dan_field_map['工作内容']}, {self.qing_dan_field_map['计算规则']}, {self.qing_dan_field_map['项目特征']}, {self.qing_dan_field_map['特征值']}) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(资源库名称, data['章节码'], data['编码'], data['名称'], data['单位'],
data['工作内容'], data['计算规则'], data['项目特征'], data['特征值'])
)
print(f" 成功处理 {len(processed_data)} 条清单子目记录")
except Exception as e:
print(f"处理清单库文件 {excel_file} 时出错: {str(e)}")
continue
print("清单库文件处理完成")
except Exception as e:
print(f"处理清单库文件时出错: {str(e)}")
self.conn.rollback()
def commit_and_close(self):
"""提交事务并关闭数据库连接"""
self.conn.commit()
self.conn.close()
print("数据库事务已提交,连接已关闭")
class CreateEmbedingData():
def __init__(self, db_path):
self.db_path = db_path
self.conn = sqlite3.connect(db_path)
self.embedding_function = XinferenceEmbeddings()
def create_ding_e_zimu_embedding(self):
"""创建定额子目名称的向量索引"""
cursor = self.conn.execute("""
SELECT dz.bm, dz.mc, dz.zyk_mc, ds.sy_fw
FROM ding_e_zimu dz
LEFT JOIN ding_e_zyk_shuxing ds ON dz.zyk_mc = ds.zyk_mc
""")
rows = cursor.fetchall()
texts = [row[1] for row in rows] # 提取描述文本
metadatas = [{"bm": row[0], "mc": row[1], "zyk_mc": row[2], "sy_fw": row[3]} for row in rows] # 添加元数据
# 创建SQLiteVSS实例
db = SQLiteVSS(
table="embeding_ding_e_zimu_name", # 向量表名
connection=None, # 复用现有连接
embedding=self.embedding_function,
db_file=self.db_path # 复用原数据库文件
)
# 分批次插入数据,每批次5000条
batch_size = 5000
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
batch_metadatas = metadatas[i:i+batch_size]
db.add_texts(texts=batch_texts, metadatas=batch_metadatas)
print(f"已插入定额子目向量索引 {i+len(batch_texts)}/{len(texts)}")
return db
def create_qd_zimu_embedding(self):
"""创建清单子目名称的向量索引"""
cursor = self.conn.execute("""
SELECT qz.bm, qz.mc, qz.zyk_mc, qs.sy_fw
FROM qd_zimu qz
LEFT JOIN qd_zyk_shuxing qs ON qz.zyk_mc = qs.zyk_mc
""")
rows = cursor.fetchall()
texts = [row[1] for row in rows] # 提取描述文本
metadatas = [{"bm": row[0], "mc": row[1], "zyk_mc": row[2], "sy_fw": row[3]} for row in rows] # 添加元数据
# 创建SQLiteVSS实例
db = SQLiteVSS(
table="embeding_qd_zimu_name", # 向量表名
connection=None, # 复用现有连接
embedding=self.embedding_function,
db_file=self.db_path # 复用原数据库文件
)
# 分批次插入数据,每批次5000条
batch_size = 5000
for i in range(0, len(texts), batch_size):
batch_texts = texts[i:i+batch_size]
batch_metadatas = metadatas[i:i+batch_size]
db.add_texts(texts=batch_texts, metadatas=batch_metadatas)
print(f"已插入清单子目向量索引 {i+len(batch_texts)}/{len(texts)}")
return db
def close(self):
"""关闭数据库连接"""
if self.conn:
self.conn.close()
def create_db():
"""主函数"""
print("开始处理定额库和清单库Excel文件...")
# 配置参数
ding_e_base_dir = f"{os.getcwd()}/data/excel/Excel版 清单定额库/定额库"
qing_dan_base_dir = f"{os.getcwd()}/data/excel/Excel版 清单定额库/清单库"
db_path = f"{os.getcwd()}/data/db/qingdan_ding_e_ku.db"
if os.path.exists(db_path):
print("数据库文件已存在, 任务结束...")
return
# 创建处理器实例
processor = ExcelToSQLiteProcessor(db_path)
try:
# 处理定额库文件
processor.process_ding_e_files(ding_e_base_dir)
# # 处理清单库文件
processor.process_qing_dan_files(qing_dan_base_dir)
# # 提交并关闭
processor.commit_and_close()
print("=" * 50)
print("所有Excel文件处理完成!数据已成功导入SQLite数据库")
print(f"数据库文件位置: {db_path}")
print("=" * 50)
# 生成向量数据
print("开始生成向量数据...")
try:
# 创建向量数据处理器实例
embedding_processor = CreateEmbedingData(db_path)
# 生成定额子目向量数据
print("正在生成定额子目向量数据...")
embedding_processor.create_ding_e_zimu_embedding()
# 生成清单子目向量数据
print("正在生成清单子目向量数据...")
embedding_processor.create_qd_zimu_embedding()
# 关闭连接
embedding_processor.close()
print("=" * 50)
print("向量数据生成完成!")
print("=" * 50)
except Exception as e:
logging.error(f"生成向量数据过程中出现错误: {str(e)}", exc_info=True)
except Exception as e:
logging.error(f"处理过程中出现错误: {str(e)}", exc_info=True)
processor.conn.rollback()
processor.conn.close()
if __name__ == "__main__":
create_db()