import re from typing import Any def split_inserts(sql_text: str) -> list[str]: parts = re.split(r"(?=INSERT INTO)", sql_text, flags=re.IGNORECASE) return [p.strip().rstrip(";") for p in parts if p.strip().upper().startswith("INSERT")] def parse_insert_columns(insert_sql: str) -> list[str]: m = re.search(r"INSERT INTO\s+[`\"]?[^`\"(\s]+[`\"]?\s*\(([^)]+)\)\s*VALUES", insert_sql, re.I) if not m: return [] return [c.strip().strip("`\"") for c in m.group(1).split(",")] def parse_values_tuple(insert_sql: str) -> list[Any]: idx = insert_sql.upper().find("VALUES") if idx < 0: return [] rest = insert_sql[idx + 6 :].strip() if not rest.startswith("("): return [] return _parse_tuple(rest) def _parse_tuple(s: str) -> list[Any]: assert s[0] == "(" i = 1 values: list[Any] = [] while i < len(s): while i < len(s) and s[i] in " \t\n\r": i += 1 if i >= len(s): break if s[i] == ")": break if i < len(s) - 3 and s[i : i + 4].upper() == "NULL": values.append(None) i += 4 while i < len(s) and s[i] in " \t": i += 1 if i < len(s) and s[i] == ",": i += 1 continue if s[i] == "'": i += 1 buf: list[str] = [] while i < len(s): ch = s[i] if ch == "\\" and i + 1 < len(s): n = s[i + 1] if n == "'": buf.append("'") i += 2 continue if n == '"': buf.append('"') i += 2 continue if n == "\\": buf.append("\\") i += 2 continue buf.append(ch) i += 1 continue if ch == "'": if i + 1 < len(s) and s[i + 1] == "'": buf.append("'") i += 2 continue i += 1 break buf.append(ch) i += 1 values.append("".join(buf)) while i < len(s) and s[i] in " \t": i += 1 if i < len(s) and s[i] == ",": i += 1 continue if s[i : i + 2] == "b'": i += 2 while i < len(s) and s[i] != "'": i += 1 i += 1 values.append(0) while i < len(s) and s[i] in " \t": i += 1 if i < len(s) and s[i] == ",": i += 1 continue m = re.match(r"-?\d+(?:\.\d+)?", s[i:]) if m: num = m.group(0) values.append(float(num) if "." in num else int(num)) i += len(num) while i < len(s) and s[i] in " \t": i += 1 if i < len(s) and s[i] == ",": i += 1 continue i += 1 return values def rows_from_sql(sql_text: str) -> list[dict[str, Any]]: rows = [] for ins in split_inserts(sql_text): cols = parse_insert_columns(ins) vals = parse_values_tuple(ins) if not cols or len(vals) != len(cols): continue rows.append(dict(zip(cols, vals))) return rows def _match_examinee(low: str) -> bool: if "examination_examinee" not in low: return False if "exam_answer" in low or "exam_examinee" in low: return False return True def _match_exam_examinee(low: str) -> bool: return "exam_examinee" in low or "考试考生" in low def _match_answer(low: str) -> bool: return "exam_answer" in low or "考生答案" in low def _match_paper_source(low: str) -> bool: return "paper_source" in low or ("试卷" in low and "item" not in low) def _match_paper_item(low: str) -> bool: return "paper_item" in low or "试题" in low def _match_exam(low: str) -> bool: if "examinee" in low or "exam_answer" in low: return False if "exam_examinee" in low: return False return "examination_exam" in low KIND_MATCHERS = { "examinee": _match_examinee, "exam_examinee": _match_exam_examinee, "answer": _match_answer, "paper_source": _match_paper_source, "paper_item": _match_paper_item, "exam": _match_exam, } def find_file_content(files: dict[str, bytes], kind: str) -> str | None: matcher = KIND_MATCHERS.get(kind) if not matcher: return None candidates: list[tuple[str, bytes]] = [] for name, data in files.items(): low = name.lower().replace("(", "(").replace(")", ")") if matcher(low): candidates.append((name, data)) if not candidates: return None candidates.sort(key=lambda x: len(x[0])) return candidates[0][1].decode("utf-8", errors="replace") def find_file_content_legacy(files: dict[str, bytes], patterns: list[str]) -> str | None: """Deprecated: use kind-based find_file_content in import_service.""" for name, data in files.items(): low = name.lower() for pat in patterns: if pat.lower() in low: return data.decode("utf-8", errors="replace") return None