592 lines
27 KiB
Python
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() |