#!/usr/bin/env python3 """ 将 docs/example.xls 中的项目数据导入 ocean 数据库 projects 表。 用法（在项目根目录或 backend 目录执行）： cd backend && PYTHONPATH=src python scripts/import_xls.py [path_to_example.xls] 默认路径：../docs/example.xls（相对 backend）或 docs/example.xls（相对项目根）。 """ import os import re import sys from datetime import datetime from pathlib import Path import xlrd # 数据库中的日期列（schema DATE 类型）；Excel 里可能填「在建中」「未开工」等文字 DATE_COLUMNS = { "sign_date", "start_date", "planned_completion_date", "actual_completion_date", "actual_warranty_refund_date", } # 质保期截止日、中标合同金额、工程款拨付方式：按字符串存储，原样导入 STRING_STORAGE_COLUMNS = {"warranty_end_date", "bid_contract_amount", "payment_method"} # 签订/开工/计划竣工/实际竣工：有 _str 字段，日期正常写 DATE+「日期正常」，非日期写 _str DATE_COLUMNS_WITH_STR = { "sign_date", "start_date", "planned_completion_date", "actual_completion_date", } DATE_PATTERN = re.compile(r"^\d{4}-\d{2}-\d{2}$") # 确保可导入 src _backend = Path(__file__).resolve().parent.parent if str(_backend) not in sys.path: sys.path.insert(0, str(_backend)) os.chdir(_backend) from src.database import get_connection, new_id # Excel 表头（中文，strip 后）-> 数据库列名（snake_case） # 依据 schema.sql、初始需求；表头可能含换行或「其中：」等前缀 EXCEL_HEADER_TO_DB = { "序号": "serial_no", "合同编号": "contract_code", "供电局项目合同编号": "power_bureau_contract_code", "项目名称": "project_name", "子项个数": "sub_item_count", "子项编码": "sub_item_code", "项目总投资（万元）": "total_investment", "中标合同金额（万元）": "bid_contract_amount", "质保金比例": "warranty_ratio", "结算金额（万元）": "settlement_amount", "总成本测算": "total_cost_estimate", "工程电压等级": "voltage_level", "工程类别": "project_category", "业主单位": "owner_unit", "业主联系人及电话": "owner_contact", "中标形式": "bid_type", "签订日期": "sign_date", "开工日期": "start_date", "计划竣工日期": "planned_completion_date", "实际竣工日期": "actual_completion_date", "质保金（万元）": "warranty_amount", "质保期截止日": "warranty_end_date", "实际退质保金日期": "actual_warranty_refund_date", "所属项目部": "project_department", "项目负责人及电话": "project_leader_contact", "工程款拨付方式": "payment_method", "总体成本（控制）": "total_cost", "是否调整（是/否）": "is_adjusted", "农民工工资（按进度计划）": "migrant_worker_plan", "农民工工资（实付）": "migrant_worker_actual", "其中：乙供材料费（控制）": "self_supply_material_control", "应付材料费（按收款比例）": "material_payable_by_ratio", "实际发生材料费": "material_actual_occurred", "实际支付材料费": "material_actual_paid", "其中：其他费用（控制）": "other_cost_control", "应付其他费": "other_payable", "实际其他费用": "other_actual", "税金": "tax", "利润（万元）": "profit", "实际利润（万元）": "actual_profit", "成本结算金额（万元）": "cost_settlement_amount", "累计进度": "cumulative_progress", "应收款（完成进度款）": "receivable_progress", "开票金额（万元）": "invoice_amount", "实际收款金额（万元）": "actual_receipt_amount", "实际收款完成率": "actual_receipt_rate", "应付款金额（万元）": "payable_amount", "实际付款金额（万元）": "actual_payment_amount", "未收款（万元）": "unreceived_amount", "实际付款完成率": "actual_payment_rate", "民工工资清欠金额（万元）": "migrant_worker_arrears", "结算后成本测算金额": "settlement_cost_estimate", "结算人工费": "settlement_labor_cost", "结算材料费": "settlement_material_cost", "结算其他费": "settlement_other_cost", "到期应结算项目个数": "due_settlement_count", "到期未完成结算个数": "overdue_settlement_count", "存在的问题": "existing_problems", "建议措施": "suggestions", "备注": "remark", } def _normalize_header(h: str) -> str: return h.strip().replace("\n", "").replace("\r", "") def _cell_value(sh: xlrd.sheet.Sheet, row: int, col: int): """取单元格值，日期转 YYYY-MM-DD，空转 None。""" try: cell = sh.cell(row, col) except IndexError: return None if cell.ctype == xlrd.XL_CELL_EMPTY: return None if cell.ctype == xlrd.XL_CELL_DATE: try: d = xlrd.xldate.xldate_as_datetime(cell.value, 0) return d.strftime("%Y-%m-%d") except Exception: return None v = cell.value if isinstance(v, float): if v != v: # NaN return None if v == int(v): v = int(v) if v == "" or (isinstance(v, str) and not v.strip()): return None if isinstance(v, str): v = v.strip() return v def _cell_value_date_and_str(sh: xlrd.sheet.Sheet, row: int, col: int) -> tuple: """专用于带 _str 的日期列。返回 (date_val, str_val)：date_val 为 YYYY-MM-DD 或 None，str_val 为「日期正常」或原文或 None。""" try: cell = sh.cell(row, col) except IndexError: return None, None if cell.ctype == xlrd.XL_CELL_EMPTY: return None, None # 日期型或数字型（Excel 序列日期） if cell.ctype == xlrd.XL_CELL_DATE: try: d = xlrd.xldate.xldate_as_datetime(cell.value, 0) return d.strftime("%Y-%m-%d"), "日期正常" except Exception: return None, None if cell.ctype == xlrd.XL_CELL_NUMBER and isinstance(cell.value, (int, float)): v = cell.value if 20000 <= v <= 50000 or (0 < v < 1): # 常见 Excel 日期范围或时间部分 try: d = xlrd.xldate.xldate_as_datetime(v, 0) return d.strftime("%Y-%m-%d"), "日期正常" except Exception: pass return None, str(int(v)) if v == int(v) else str(v) # 文本 v = cell.value if v is None or (isinstance(v, str) and not v.strip()): return None, None s = (v if isinstance(v, str) else str(v)).strip() if DATE_PATTERN.match(s): return s, "日期正常" return None, s def _cell_value_raw_str(sh: xlrd.sheet.Sheet, row: int, col: int): """取单元格原值并转为字符串（用于 warranty_end_date、bid_contract_amount 等字符串存储列）。""" try: cell = sh.cell(row, col) except IndexError: return None if cell.ctype == xlrd.XL_CELL_EMPTY: return None v = cell.value if v is None: return None s = (str(v) if not isinstance(v, str) else v).strip() return s if s else None def _row_to_db_dict(sh: xlrd.sheet.Sheet, header_to_col: dict, row_idx: int) -> dict: """将一行转为 DB 列名 -> 值的字典。header_to_col: 表头(已 normalize) -> 列索引。""" row_dict = {} for header, col in header_to_col.items(): db_col = EXCEL_HEADER_TO_DB.get(header) if not db_col: continue if db_col in DATE_COLUMNS_WITH_STR: date_val, str_val = _cell_value_date_and_str(sh, row_idx, col) if date_val is not None: row_dict[db_col] = date_val row_dict[db_col + "_str"] = "日期正常" elif str_val is not None and str_val.strip(): row_dict[db_col + "_str"] = str_val continue if db_col in STRING_STORAGE_COLUMNS: val = _cell_value_raw_str(sh, row_idx, col) if val is not None: row_dict[db_col] = val continue if db_col == "total_cost": val = _cell_value(sh, row_idx, col) if val is not None: if isinstance(val, (int, float)) and (not isinstance(val, float) or not (val != val)): row_dict[db_col] = float(val) elif isinstance(val, str): s = val.strip() try: row_dict[db_col] = float(s) except (ValueError, TypeError): row_dict[db_col] = 0 row_dict["_total_cost_text"] = s else: row_dict[db_col] = 0 continue val = _cell_value(sh, row_idx, col) if val is None: continue if db_col in DATE_COLUMNS and isinstance(val, str) and not DATE_PATTERN.match(val.strip()): continue row_dict[db_col] = val if "_total_cost_text" in row_dict: prefix = "总体成本（文字）：" + row_dict["_total_cost_text"] + "\n" row_dict["remark"] = prefix + (row_dict.get("remark") or "") del row_dict["_total_cost_text"] return row_dict def main(): if len(sys.argv) >= 2: xls_path = Path(sys.argv[1]).resolve() else: for p in [_backend.parent / "docs" / "example.xls", _backend / "docs" / "example.xls"]: if p.exists(): xls_path = p break else: print("未找到 example.xls，请指定路径：python scripts/import_xls.py ") sys.exit(1) if not xls_path.exists(): print(f"文件不存在：{xls_path}") sys.exit(1) wb = xlrd.open_workbook(str(xls_path)) sh = wb.sheet_by_index(0) header_row_idx = 2 data_start_idx = 5 raw_headers = [str(sh.cell_value(header_row_idx, j)) for j in range(sh.ncols)] # header_to_col: 归一化表头 -> 列索引（取首次出现） header_to_col = {} for j, h in enumerate(raw_headers): nh = _normalize_header(h) if nh and nh not in header_to_col: header_to_col[nh] = j contract_code_col = header_to_col.get("合同编号") project_name_col = header_to_col.get("项目名称") if contract_code_col is None: print("未找到列「合同编号」，请检查 Excel 表头。") sys.exit(1) inserted = 0 skipped = 0 with get_connection() as conn: cur = conn.cursor() for r in range(data_start_idx, sh.nrows): row_dict = _row_to_db_dict(sh, header_to_col, r) contract_code = _cell_value(sh, r, contract_code_col) if contract_code_col is not None else row_dict.get("contract_code") project_name = _cell_value(sh, r, project_name_col) if project_name_col is not None else row_dict.get("project_name") if contract_code is None or (isinstance(contract_code, str) and not contract_code.strip()): skipped += 1 continue if project_name is None or (isinstance(project_name, str) and not project_name.strip()): project_name = str(contract_code) if contract_code else "未命名" row_dict["contract_code"] = contract_code row_dict["project_name"] = project_name pid = new_id() now = datetime.utcnow() row_dict["id"] = pid row_dict["progress"] = (str(row_dict.get("cumulative_progress") or "")[:32]) if row_dict.get("cumulative_progress") is not None else "" row_dict["cost"] = "" row_dict["created_at"] = now row_dict["updated_at"] = now cols = list(row_dict.keys()) vals = [row_dict[k] for k in cols] placeholders = ", ".join(["%s"] * len(cols)) try: cur.execute( "INSERT INTO projects (" + ", ".join(cols) + ") VALUES (" + placeholders + ")", vals, ) inserted += 1 except Exception as e: print(f"Row {r + 1} 跳过（{e}）") skipped += 1 cur.close() print(f"导入完成：成功 {inserted} 条，跳过 {skipped} 条。") if __name__ == "__main__": main()