ddl_parser.py 33 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879
  1. from __future__ import annotations
  2. import io
  3. import json
  4. import logging
  5. import re
  6. import time
  7. from typing import Any
  8. import requests
  9. from app.core.llm.deepseek_client import (
  10. get_llm_api_key,
  11. get_llm_chat_completions_url,
  12. get_llm_model,
  13. )
  14. logger = logging.getLogger(__name__)
  15. class DDLParser:
  16. def __init__(self, api_key=None, timeout=60, max_retries=3):
  17. """
  18. 初始化DDL解析器
  19. 参数:
  20. api_key: LLM API密钥,如果未提供,将从应用配置或环境变量中获取
  21. timeout: API请求超时时间(秒),默认60秒
  22. max_retries: 最大重试次数,默认3次
  23. """
  24. # 如果在Flask应用上下文中,则从应用配置获取参数
  25. self.api_key = api_key or get_llm_api_key()
  26. self.chat_completions_url = get_llm_chat_completions_url()
  27. self.model_name = get_llm_model()
  28. self.timeout = timeout
  29. self.max_retries = max_retries
  30. if not self.api_key:
  31. logger.error(
  32. "DeepSeek API Key 未配置,请在 /etc/dataops-platform/dataops.env "
  33. "中设置 DEEPSEEK_API_KEY 后重启服务"
  34. )
  35. self.headers = {
  36. "Authorization": f"Bearer {self.api_key}",
  37. "Content-Type": "application/json",
  38. }
  39. def _make_llm_request(self, payload, operation_name="LLM请求"):
  40. """
  41. 发送LLM请求,支持自动重试
  42. 参数:
  43. payload: 请求payload
  44. operation_name: 操作名称,用于日志
  45. 返回:
  46. API响应结果
  47. """
  48. last_error = None
  49. if not self.api_key:
  50. logger.error(f"{operation_name} 跳过: DeepSeek API Key 未配置")
  51. return None
  52. for attempt in range(self.max_retries):
  53. try:
  54. if attempt > 0:
  55. wait_time = 2**attempt # 指数退避: 2, 4, 8秒
  56. logger.info(
  57. f"{operation_name} 第{attempt + 1}次重试,等待{wait_time}秒..."
  58. )
  59. time.sleep(wait_time)
  60. logger.info(
  61. f"{operation_name} 尝试 {attempt + 1}/{self.max_retries},超时时间: {self.timeout}秒"
  62. )
  63. response = requests.post(
  64. self.chat_completions_url,
  65. headers=self.headers,
  66. json=payload,
  67. timeout=self.timeout,
  68. )
  69. response.raise_for_status()
  70. result = response.json()
  71. logger.info(f"{operation_name} 成功")
  72. return result
  73. except requests.Timeout as e:
  74. last_error = f"请求超时(超过{self.timeout}秒): {str(e)}"
  75. logger.warning(f"{operation_name} 超时: {str(e)}")
  76. except requests.RequestException as e:
  77. status_code = getattr(getattr(e, "response", None), "status_code", None)
  78. if status_code == 401:
  79. last_error = (
  80. "DeepSeek API 鉴权失败(401),请检查 DEEPSEEK_API_KEY 是否正确、"
  81. "是否已开通余额,并重载服务"
  82. )
  83. else:
  84. last_error = f"API请求失败: {str(e)}"
  85. logger.warning(f"{operation_name} 失败: {last_error}")
  86. except Exception as e:
  87. last_error = f"未知错误: {str(e)}"
  88. logger.error(f"{operation_name} 异常: {str(e)}")
  89. break # 对于非网络错误,不重试
  90. # 所有重试都失败
  91. logger.error(f"{operation_name} 在{self.max_retries}次尝试后失败: {last_error}")
  92. return None
  93. @staticmethod
  94. def _split_sql_identifier_list(section: str) -> list[str]:
  95. names: list[str] = []
  96. for part in re.split(r",\s*", section.strip()):
  97. cleaned = part.strip().strip('"').strip("'")
  98. if cleaned:
  99. names.append(cleaned)
  100. return names
  101. @staticmethod
  102. def _build_column_defs(column_names: list[str]) -> list[dict[str, str]]:
  103. return [
  104. {
  105. "name_zh": "",
  106. "name_en": name,
  107. "data_type": "VARCHAR(255)",
  108. "is_primary": "否",
  109. "comment": "",
  110. "nullable": "是",
  111. }
  112. for name in column_names
  113. ]
  114. def _parse_create_views(self, sql_content: str) -> list[dict]:
  115. view_pattern = re.compile(
  116. r'CREATE\s+(?:OR\s+REPLACE\s+)?(?:\w+\s+)*VIEW\s+'
  117. r'(?:"?(?:[\w$#]+)"?\.)?"?([\w$#]+)"?\s*\(([^)]+)\)',
  118. re.IGNORECASE | re.DOTALL,
  119. )
  120. results: list[dict] = []
  121. for match in view_pattern.finditer(sql_content):
  122. table_name = match.group(1)
  123. column_names = self._split_sql_identifier_list(match.group(2))
  124. if not column_names:
  125. continue
  126. results.append(
  127. {
  128. "table_info": {
  129. "name_zh": "",
  130. "name_en": table_name,
  131. },
  132. "columns": self._build_column_defs(column_names),
  133. }
  134. )
  135. return results
  136. def _parse_create_tables(self, sql_content: str) -> list[dict]:
  137. table_pattern = re.compile(
  138. r'CREATE\s+TABLE\s+(?:"?(?:[\w$#]+)"?\.)?"?([\w$#]+)"?\s*\(',
  139. re.IGNORECASE | re.DOTALL,
  140. )
  141. results: list[dict] = []
  142. for match in table_pattern.finditer(sql_content):
  143. start = match.end()
  144. depth = 1
  145. index = start
  146. while index < len(sql_content) and depth > 0:
  147. char = sql_content[index]
  148. if char == "(":
  149. depth += 1
  150. elif char == ")":
  151. depth -= 1
  152. index += 1
  153. if depth != 0:
  154. continue
  155. body = sql_content[start : index - 1]
  156. columns: list[dict[str, str]] = []
  157. for line in body.splitlines():
  158. line = line.strip().rstrip(",")
  159. if not line or line.upper().startswith(
  160. ("CONSTRAINT", "PRIMARY", "UNIQUE", "FOREIGN", "CHECK", "INDEX")
  161. ):
  162. continue
  163. col_match = re.match(
  164. r'^"?([\w$#]+)"?\s+([A-Za-z][\w$#()]*(?:\([^)]*\))?)',
  165. line,
  166. re.IGNORECASE,
  167. )
  168. if not col_match:
  169. continue
  170. col_name = col_match.group(1)
  171. data_type = col_match.group(2).upper()
  172. upper_line = line.upper()
  173. columns.append(
  174. {
  175. "name_zh": "",
  176. "name_en": col_name,
  177. "data_type": data_type,
  178. "is_primary": "是" if "PRIMARY KEY" in upper_line else "否",
  179. "comment": "",
  180. "nullable": "否" if "NOT NULL" in upper_line else "是",
  181. }
  182. )
  183. if columns:
  184. results.append(
  185. {
  186. "table_info": {
  187. "name_zh": "",
  188. "name_en": match.group(1),
  189. },
  190. "columns": columns,
  191. }
  192. )
  193. return results
  194. def _parse_sql_ddl_fallback(self, sql_content: str) -> list[dict]:
  195. """Parse CREATE VIEW / CREATE TABLE locally when LLM output is empty or invalid."""
  196. results = self._parse_create_views(sql_content)
  197. if results:
  198. return results
  199. return self._parse_create_tables(sql_content)
  200. @staticmethod
  201. def normalize_ddl_parse_result(raw: Any) -> list[dict]:
  202. """Normalize LLM or legacy parser output into standard table list."""
  203. if raw is None:
  204. return []
  205. if isinstance(raw, list):
  206. return [
  207. item
  208. for item in raw
  209. if isinstance(item, dict) and isinstance(item.get("table_info"), dict)
  210. ]
  211. if not isinstance(raw, dict):
  212. return []
  213. if raw.get("code") == 500 and "table_info" not in raw:
  214. return []
  215. if "table_info" in raw:
  216. return [raw]
  217. converted: list[dict] = []
  218. for table_name, table_data in raw.items():
  219. if not isinstance(table_data, dict):
  220. continue
  221. if "table_info" in table_data:
  222. converted.append(table_data)
  223. continue
  224. columns = table_data.get("columns")
  225. if isinstance(columns, list):
  226. converted.append(
  227. {
  228. "table_info": {
  229. "name_zh": table_data.get("name_zh", ""),
  230. "name_en": table_data.get("name_en", table_name),
  231. },
  232. "columns": columns,
  233. }
  234. )
  235. return converted
  236. def _parse_ddl_with_llm(self, sql_content: str) -> Any:
  237. prompt = self._optimize_ddl_prompt()
  238. payload = {
  239. "model": self.model_name,
  240. "messages": [
  241. {
  242. "role": "system",
  243. "content": (
  244. "你是一个专业的SQL DDL语句解析专家,擅长从DDL建表语句和"
  245. "CREATE VIEW视图定义中提取表结构信息并转换为结构化的JSON格式。"
  246. ),
  247. },
  248. {"role": "user", "content": f"{prompt}\n\n{sql_content}"},
  249. ],
  250. }
  251. result = self._make_llm_request(payload, "DDL解析")
  252. if not result:
  253. return {
  254. "code": 500,
  255. "message": f"API请求失败: 在{self.max_retries}次尝试后仍然失败",
  256. }
  257. if "choices" not in result or not result["choices"]:
  258. return {
  259. "code": 500,
  260. "message": "无法获取有效响应",
  261. "original_response": result,
  262. }
  263. content = result["choices"][0]["message"]["content"]
  264. try:
  265. json_match = re.search(r"```json\s*([\s\S]*?)\s*```", content)
  266. json_content = json_match.group(1) if json_match else content
  267. return json.loads(json_content)
  268. except json.JSONDecodeError as exc:
  269. return {
  270. "code": 500,
  271. "message": f"无法解析返回的JSON: {str(exc)}",
  272. "original_response": content,
  273. }
  274. def parse_ddl(self, sql_content):
  275. """
  276. 解析DDL语句,返回标准化的结构
  277. 参数:
  278. sql_content: 要解析的DDL语句
  279. 返回:
  280. 标准表结构数组;优先本地解析,复杂语句再调用 LLM
  281. """
  282. try:
  283. fallback_list = self._parse_sql_ddl_fallback(sql_content)
  284. if fallback_list:
  285. logger.info(
  286. f"DDL 本地SQL解析成功,识别 {len(fallback_list)} 个表/视图"
  287. )
  288. return fallback_list
  289. llm_raw = self._parse_ddl_with_llm(sql_content)
  290. ddl_list = self.normalize_ddl_parse_result(llm_raw)
  291. if ddl_list:
  292. logger.info(f"DDL LLM解析成功,识别 {len(ddl_list)} 个表/视图")
  293. return ddl_list
  294. if isinstance(llm_raw, dict) and llm_raw.get("message"):
  295. return llm_raw
  296. return []
  297. except Exception as e:
  298. logger.error(f"DDL解析异常: {str(e)}")
  299. fallback_list = self._parse_sql_ddl_fallback(sql_content)
  300. if fallback_list:
  301. logger.info(
  302. f"DDL解析异常后使用本地SQL解析,识别 {len(fallback_list)} 个表/视图"
  303. )
  304. return fallback_list
  305. return {"code": 500, "message": f"解析失败: {str(e)}"}
  306. def parse_db_conn_str(self, conn_str):
  307. """
  308. 解析数据库连接字符串
  309. 参数:
  310. conn_str: 要解析的数据库连接字符串
  311. 返回:
  312. 解析结果的JSON对象
  313. """
  314. prompt = self._optimize_connstr_parse_prompt()
  315. payload = {
  316. "model": self.model_name,
  317. "messages": [
  318. {
  319. "role": "system",
  320. "content": "你是一个专业的数据库连接字符串解析专家,擅长解析各种数据库的连接字符串并提取关键信息。",
  321. },
  322. {"role": "user", "content": f"{prompt}\n\n{conn_str}"},
  323. ],
  324. }
  325. try:
  326. result = self._make_llm_request(payload, "连接字符串解析")
  327. if not result:
  328. return {
  329. "code": 500,
  330. "message": f"API请求失败: 在{self.max_retries}次尝试后仍然失败",
  331. }
  332. if "choices" in result and len(result["choices"]) > 0:
  333. content = result["choices"][0]["message"]["content"]
  334. try:
  335. json_match = re.search(r"```json\s*([\s\S]*?)\s*```", content)
  336. if json_match:
  337. json_content = json_match.group(1)
  338. else:
  339. json_content = content
  340. parsed_result = json.loads(json_content)
  341. return parsed_result
  342. except json.JSONDecodeError as e:
  343. return {
  344. "code": 500,
  345. "message": f"无法解析返回的JSON: {str(e)}",
  346. "original_response": content,
  347. }
  348. return {
  349. "code": 500,
  350. "message": "无法获取有效响应",
  351. "original_response": result,
  352. }
  353. except Exception as e:
  354. logger.error(f"连接字符串解析异常: {str(e)}")
  355. return {"code": 500, "message": f"解析失败: {str(e)}"}
  356. def _optimize_ddl_prompt(self):
  357. """返回优化后的提示词模板"""
  358. return """
  359. 请解析以下DDL建表语句或CREATE VIEW视图定义,并按照指定的JSON格式返回结果:
  360. 规则说明:
  361. 1. 从DDL语句中识别所有表和视图,可能会有多个对象。将所有对象放在一个数组中返回。
  362. 2. CREATE VIEW 视图的列名来自视图定义括号中的列清单;若无中文注释,name_zh 留空。
  363. 3. 表的英文名称(name_en)使用原始大小写,不要转换为小写。
  364. 3. 表的中文名称(name_zh)提取规则:
  365. - 优先从COMMENT ON TABLE语句中提取
  366. - 如果没有注释,则name_zh为空字符串
  367. - 中文名称中不要出现标点符号、"主键"、"外键"、"索引"等字样
  368. 4. 对于每个表,提取所有字段信息到columns数组中,每个字段包含:
  369. - name_zh: 字段中文名称(从COMMENT ON COLUMN提取,如果没有注释则翻译英文名,如果是无意义缩写则为空)
  370. - name_en: 字段英文名称(保持原始大小写)
  371. - data_type: 数据类型(包含长度信息,如VARCHAR(22))
  372. - is_primary: 是否主键("是"或"否",从PRIMARY KEY约束判断)
  373. - comment: 注释内容(从COMMENT ON COLUMN提取完整注释,如果没有则为空字符串)
  374. - nullable: 是否可为空("是"或"否",从NOT NULL约束判断,默认为"是")
  375. 5. 中文字段名不要出现逗号、"主键"、"外键"、"索引"等字样。
  376. 6. 返回格式(使用数组支持多表):
  377. [
  378. {
  379. "table_info": {
  380. "name_zh": "科室对照表",
  381. "name_en": "TB_JC_KSDZB"
  382. },
  383. "columns": [
  384. {
  385. "name_zh": "医疗机构代码",
  386. "name_en": "YLJGDM",
  387. "data_type": "VARCHAR(22)",
  388. "is_primary": "是",
  389. "comment": "医疗机构代码,复合主键",
  390. "nullable": "否"
  391. },
  392. {
  393. "name_zh": "HIS科室代码",
  394. "name_en": "HISKSDM",
  395. "data_type": "CHAR(20)",
  396. "is_primary": "是",
  397. "comment": "HIS科室代码,主键、唯一",
  398. "nullable": "否"
  399. },
  400. {
  401. "name_zh": "HIS科室名称",
  402. "name_en": "HISKSMC",
  403. "data_type": "CHAR(20)",
  404. "is_primary": "否",
  405. "comment": "HIS科室名称",
  406. "nullable": "否"
  407. }
  408. ]
  409. }
  410. ]
  411. 注意:
  412. - 如果只有一个表,也要返回数组格式:[{table_info: {...}, columns: [...]}]
  413. - 如果有多个表,数组中包含多个元素:[{表1}, {表2}, {表3}]
  414. 请仅返回JSON格式结果,不要包含任何其他解释文字。
  415. """
  416. def _optimize_ddl_source_prompt(self):
  417. """返回优化后的提示词模板"""
  418. return """
  419. 请解析以下DDL建表语句,并按照指定的JSON格式返回结果:
  420. 规则说明:
  421. 1. 从DDL语句中识别所有表名,并在data对象中为每个表创建条目,表名请使用小写,可能会有多个表。
  422. 2. 对于每个表,提取所有字段信息,包括名称、数据类型和注释。
  423. - 中文表名中不要出现标点符号
  424. 3. 字段中文名称(name_zh)的确定规则:
  425. - 如有COMMENT注释,直接使用注释内容
  426. - 如无注释但字段名有明确含义,将英文名翻译为中文
  427. - 如字段名是无意义的拼音缩写,则name_zh为空字符串
  428. - 字段名中不要出现逗号,以及"主键"、"外键"、"索引"等字样
  429. 4. 所有的表的定义信息,请放在tables对象中, tables对象的key为表名,value为表的定义信息。这里可能会有多个表,请一一识别。
  430. 5. data_source对象,请放在data_source标签中,它与tables对象同级。
  431. 6. 数据库连接串处理:
  432. - 将连接串识别后并拆解为:主机名/IP地址、端口、数据库名称、用户名、密码。
  433. - 根据连接串格式识别数据库类型,数据库类型请使用小写,参考例子,如 mysql/postgresql/sqlserver/oracle/db2/sybase
  434. - data_source.name_en格式为: "{数据库名称}_{hostname或ip地址}_{端口}_{数据库用户名}",如某个元素无法识别,则跳过不添加.
  435. - data_source.name_zh留空.
  436. - 无法确定数据库类型时,type设为"unknown"
  437. - 如果从ddl中没有识别到数据库连接串,则json不返回"data_source"标签
  438. - 除了database,password,username,name_en,host,port,type,name_zh 之外,连接串的其它字段放在param属性中。
  439. 7. 参考格式如下:
  440. {
  441. "tables": {
  442. "users": { //表名
  443. "name_zh": "用户表", //表的中文名,来自于COMMENT注释或LLM翻译,如果无法确定,则name_zh为空字符串
  444. "schema": "public",
  445. "meta": [{
  446. "name_en": "id",
  447. "data_type": "integer",
  448. "name_zh": "用户ID"
  449. },
  450. {
  451. "name_en": "username",
  452. "data_type": "varchar",
  453. "name_zh": "用户名"
  454. }
  455. ]
  456. }
  457. },
  458. "data_source": [{
  459. "name_en": "mydatabase_10.52.31.104_5432_myuser", //{数据库名称}_{hostname或ip地址}_{端口}_{数据库用户名}
  460. "name_zh": "", //如果没有注释,这里留空
  461. "type": "postgresql",
  462. "host": "10.52.31.104",
  463. "port": 5432,
  464. "database": "mydatabase",
  465. "username": "myuser",
  466. "password": "mypassword",
  467. "param": "useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
  468. }]
  469. }
  470. 请仅返回JSON格式结果,不要包含任何其他解释文字。
  471. """
  472. def _optimize_connstr_parse_prompt(self):
  473. """返回优化后的连接字符串解析提示词模板"""
  474. return """
  475. 请解析以下数据库连接字符串,并按照指定的JSON格式返回结果:
  476. 规则说明:
  477. 1. 将连接串识别后并拆解为:主机名/IP地址、端口、数据库名称、用户名、密码。
  478. 2. 根据连接串格式识别数据库类型,数据库类型请使用小写,如 mysql/postgresql/sqlserver/oracle/db2/sybase
  479. 3. data_source.name_en格式为: "{数据库名称}_{hostname或ip地址}_{端口}_{数据库用户名}",如某个元素无法识别,则跳过不添加
  480. 4. data_source.name_zh留空
  481. 5. 无法确定数据库类型时,type设为"unknown"
  482. 6. 除了database,password,username,name_en,host,port,type,name_zh 之外,连接串的其它字段放在param属性中
  483. 返回格式示例:
  484. {
  485. "data_source": {
  486. "name_en": "mydatabase_10.52.31.104_5432_myuser",
  487. "name_zh": "",
  488. "type": "postgresql",
  489. "host": "10.52.31.104",
  490. "port": 5432,
  491. "database": "mydatabase",
  492. "username": "myuser",
  493. "password": "mypassword",
  494. "param": "useUnicode=true&characterEncoding=utf8&serverTimezone=UTC"
  495. }
  496. }
  497. 请仅返回JSON格式结果,不要包含任何其他解释文字。
  498. """
  499. def _optimize_connstr_valid_prompt(self):
  500. """返回优化后的连接字符串验证提示词模板"""
  501. return """
  502. 请验证以下数据库连接信息是否符合规则:
  503. 规则说明:
  504. 1. 必填字段检查:
  505. - database: 数据库名称,不能为空,符合数据库名称的命名规范。
  506. - name_en: 格式必须为 "{数据库名称}_{hostname或ip地址}_{端口}_{数据库用户名}"
  507. - host: 主机名或IP地址,不能为空
  508. - port: 端口号,必须为数字
  509. - type: 数据库类型,必须为以下之一:mysql/postgresql/sqlserver/oracle/db2/sybase
  510. - username: 用户名,不能为空,名称中间不能有空格。
  511. 2. 字段格式检查:
  512. - en_name中的各个部分必须与对应的字段值匹配
  513. - port必须是有效的端口号(1-65535)
  514. - type必须是小写的数据库类型名称
  515. - param中的参数格式必须正确(key=value格式)
  516. 3. 可选字段:
  517. - password: 密码(可选)
  518. - name: 中文名称(可选)
  519. - desc: 描述(可选)
  520. 请检查提供的连接信息是否符合以上规则,如果符合则返回"success",否则返回"failure"。
  521. 请仅返回"success"或"failure",不要包含任何其他解释文字。
  522. """
  523. def valid_db_conn_str(self, conn_str):
  524. """
  525. 验证数据库连接字符串是否符合规则
  526. 参数:
  527. conn_str: 要验证的数据库连接信息(JSON格式)
  528. 返回:
  529. "success" 或 "failure"
  530. """
  531. prompt = self._optimize_connstr_valid_prompt()
  532. payload = {
  533. "model": self.model_name,
  534. "messages": [
  535. {
  536. "role": "system",
  537. "content": "你是一个专业的数据库连接信息验证专家,擅长验证数据库连接信息的完整性和正确性。",
  538. },
  539. {
  540. "role": "user",
  541. "content": f"{prompt}\n\n{json.dumps(conn_str, ensure_ascii=False)}",
  542. },
  543. ],
  544. }
  545. try:
  546. result = self._make_llm_request(payload, "连接字符串验证")
  547. if not result:
  548. logger.error(
  549. f"连接字符串验证失败: 在{self.max_retries}次尝试后仍然失败"
  550. )
  551. return "failure"
  552. if "choices" in result and len(result["choices"]) > 0:
  553. content = result["choices"][0]["message"]["content"].strip().lower()
  554. return "success" if content == "success" else "failure"
  555. return "failure"
  556. except Exception as e:
  557. logger.error(f"LLM 验证数据库连接字符串失败: {str(e)}")
  558. return "failure"
  559. def parse_excel_content(self, file_content: bytes) -> list[dict[str, Any]]:
  560. """
  561. 解析 Excel 文件内容,提取数据表定义信息
  562. Args:
  563. file_content: Excel 文件的二进制内容
  564. Returns:
  565. 解析后的表结构列表
  566. """
  567. try:
  568. import pandas as pd
  569. # 读取 Excel 文件的所有 sheet
  570. excel_file = io.BytesIO(file_content)
  571. xl = pd.ExcelFile(excel_file)
  572. # 将所有 sheet 的内容转换为文本
  573. all_content = []
  574. for sheet_name in xl.sheet_names:
  575. df = pd.read_excel(xl, sheet_name=sheet_name)
  576. # 将 DataFrame 转换为 markdown 表格格式
  577. sheet_content = f"## Sheet: {sheet_name}\n"
  578. sheet_content += df.to_markdown(index=False)
  579. all_content.append(sheet_content)
  580. combined_content = "\n\n".join(all_content)
  581. logger.info(f"Excel 文件解析完成,共 {len(xl.sheet_names)} 个 sheet")
  582. # 使用 LLM 解析表结构
  583. return self._parse_document_content(combined_content, "Excel")
  584. except Exception as e:
  585. logger.error(f"Excel 文件解析失败: {str(e)}")
  586. raise ValueError(f"Excel 文件解析失败: {str(e)}") from e
  587. def parse_word_content(self, file_content: bytes) -> list[dict[str, Any]]:
  588. """
  589. 解析 Word 文件内容,提取数据表定义信息
  590. Args:
  591. file_content: Word 文件的二进制内容
  592. Returns:
  593. 解析后的表结构列表
  594. """
  595. try:
  596. from docx import Document
  597. # 读取 Word 文件
  598. doc = Document(io.BytesIO(file_content))
  599. # 提取所有段落文本
  600. paragraphs = [para.text for para in doc.paragraphs if para.text.strip()]
  601. # 提取所有表格
  602. tables_content = []
  603. for table_idx, table in enumerate(doc.tables):
  604. table_text = f"\n### 表格 {table_idx + 1}:\n"
  605. for row in table.rows:
  606. row_text = " | ".join(cell.text.strip() for cell in row.cells)
  607. table_text += row_text + "\n"
  608. tables_content.append(table_text)
  609. # 组合内容
  610. combined_content = "\n".join(paragraphs)
  611. if tables_content:
  612. combined_content += "\n\n## 文档中的表格:\n" + "\n".join(tables_content)
  613. logger.info(
  614. f"Word 文件解析完成,共 {len(paragraphs)} 个段落,{len(doc.tables)} 个表格"
  615. )
  616. # 使用 LLM 解析表结构
  617. return self._parse_document_content(combined_content, "Word")
  618. except Exception as e:
  619. logger.error(f"Word 文件解析失败: {str(e)}")
  620. raise ValueError(f"Word 文件解析失败: {str(e)}") from e
  621. def parse_pdf_content(self, file_content: bytes) -> list[dict[str, Any]]:
  622. """
  623. 解析 PDF 文件内容,提取数据表定义信息
  624. Args:
  625. file_content: PDF 文件的二进制内容
  626. Returns:
  627. 解析后的表结构列表
  628. """
  629. try:
  630. import pdfplumber
  631. # 读取 PDF 文件
  632. pdf = pdfplumber.open(io.BytesIO(file_content))
  633. all_content = []
  634. for page_num, page in enumerate(pdf.pages):
  635. page_text = f"## 第 {page_num + 1} 页:\n"
  636. # 提取页面文本
  637. text = page.extract_text()
  638. if text:
  639. page_text += text + "\n"
  640. # 提取页面中的表格
  641. tables = page.extract_tables()
  642. for table_idx, table in enumerate(tables):
  643. page_text += f"\n### 表格 {table_idx + 1}:\n"
  644. for row in table:
  645. row_text = " | ".join(str(cell) if cell else "" for cell in row)
  646. page_text += row_text + "\n"
  647. all_content.append(page_text)
  648. pdf.close()
  649. combined_content = "\n\n".join(all_content)
  650. logger.info(f"PDF 文件解析完成,共 {len(pdf.pages)} 页")
  651. # 使用 LLM 解析表结构
  652. return self._parse_document_content(combined_content, "PDF")
  653. except Exception as e:
  654. logger.error(f"PDF 文件解析失败: {str(e)}")
  655. raise ValueError(f"PDF 文件解析失败: {str(e)}") from e
  656. def _parse_document_content(
  657. self, content: str, file_type: str
  658. ) -> list[dict[str, Any]]:
  659. """
  660. 使用 LLM 解析文档内容,提取数据表定义信息
  661. Args:
  662. content: 文档的文本内容
  663. file_type: 文件类型(用于日志记录)
  664. Returns:
  665. 解析后的表结构列表
  666. """
  667. prompt = self._get_document_parse_prompt()
  668. payload = {
  669. "model": self.model_name,
  670. "messages": [
  671. {
  672. "role": "system",
  673. "content": "你是一个专业的数据表结构解析专家,擅长从各种文档中识别和提取数据表定义信息并转换为结构化的JSON格式。",
  674. },
  675. {"role": "user", "content": f"{prompt}\n\n{content}"},
  676. ],
  677. }
  678. try:
  679. result = self._make_llm_request(payload, f"{file_type}文档解析")
  680. if not result:
  681. raise ValueError(f"API请求失败: 在{self.max_retries}次尝试后仍然失败")
  682. if "choices" in result and len(result["choices"]) > 0:
  683. response_content = result["choices"][0]["message"]["content"]
  684. try:
  685. json_match = re.search(
  686. r"```json\s*([\s\S]*?)\s*```", response_content
  687. )
  688. if json_match:
  689. json_content = json_match.group(1)
  690. else:
  691. json_content = response_content
  692. parsed_result = json.loads(json_content)
  693. # 确保返回的是列表格式
  694. if isinstance(parsed_result, dict):
  695. parsed_result = [parsed_result]
  696. return parsed_result
  697. except json.JSONDecodeError as e:
  698. raise ValueError(f"无法解析返回的JSON: {str(e)}") from e
  699. raise ValueError("无法获取有效响应")
  700. except Exception as e:
  701. logger.error(f"{file_type}文档解析异常: {str(e)}")
  702. raise
  703. def _get_document_parse_prompt(self) -> str:
  704. """返回文档解析的提示词模板"""
  705. return """
  706. 请从以下文档内容中识别并提取所有数据表的定义信息,按照指定的JSON格式返回结果。
  707. 规则说明:
  708. 1. 仔细阅读文档内容,识别所有描述数据表结构的部分。
  709. 2. 一个文档可能包含一个或多个数据表的定义,请将所有表放在一个JSON数组中返回。
  710. 3. 表的英文名称(name_en):
  711. - 如果文档中有英文表名,使用原始大小写
  712. - 如果没有英文名,尝试根据中文名翻译或生成合适的英文名
  713. 4. 表的中文名称(name_zh):
  714. - 从文档中提取表的中文名称或描述
  715. - 如果没有明确的中文名,根据内容推断
  716. 5. 对于每个表,提取所有字段信息到columns数组中,每个字段包含:
  717. - name_zh: 字段中文名称
  718. - name_en: 字段英文名称(如果没有,根据中文名翻译)
  719. - data_type: 数据类型(如VARCHAR(255)、INTEGER、DATE等,如果文档未指定则根据字段用途推断)
  720. - is_primary: 是否主键("是"或"否")
  721. - comment: 字段说明或注释
  722. - nullable: 是否可为空("是"或"否",如果文档未指定默认为"是")
  723. 6. 返回格式(必须是JSON数组):
  724. [
  725. {
  726. "table_info": {
  727. "name_zh": "用户信息表",
  728. "name_en": "user_info"
  729. },
  730. "columns": [
  731. {
  732. "name_zh": "用户ID",
  733. "name_en": "user_id",
  734. "data_type": "INTEGER",
  735. "is_primary": "是",
  736. "comment": "用户唯一标识",
  737. "nullable": "否"
  738. },
  739. {
  740. "name_zh": "用户名",
  741. "name_en": "username",
  742. "data_type": "VARCHAR(50)",
  743. "is_primary": "否",
  744. "comment": "用户登录名",
  745. "nullable": "否"
  746. }
  747. ]
  748. }
  749. ]
  750. 注意:
  751. - 即使只识别到一个表,也必须返回数组格式:[{table_info: {...}, columns: [...]}]
  752. - 如果文档中没有找到任何数据表定义,返回空数组:[]
  753. - 请仅返回JSON格式结果,不要包含任何其他解释文字。
  754. """