niuniu/JhHardwareWRS_BackPoint
4
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 校验报工归属表结构

Browse Source
This commit is contained in:
souplearn 2026-07-25 02:43:17 +08:00
parent 4a7e32a0c5
commit c64229ab78
1 changed files with 110 additions and 21 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

131
scripts/migrate_report_allocations.py
View File

@ -13,23 +13,36 @@ from app.database import engine # noqa: E402
TABLE_NAME = "production_report_allocations"
IDENTIFIER_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")
REQUIRED_COLUMNS = {
"id",
"report_id",
"report_item_id",
"attendance_point_name",
"employee_phone",
"allocation_date",
"day_minutes",
"overtime_minutes",
"night_minutes",
"effective_minutes",
"good_qty",
"defect_qty",
"scrap_qty",
"changeover_count",
"reference_wage",
"created_at",
"updated_at",
"id": {"column_type": "bigint", "is_nullable": "NO", "extra_contains": ("auto_increment",)},
"report_id": {"column_type": "bigint", "is_nullable": "NO"},
"report_item_id": {"column_type": "bigint", "is_nullable": "YES"},
"attendance_point_name": {
"column_type": "varchar(128)",
"is_nullable": "NO",
"default": "empty_string",
},
"employee_phone": {"column_type": "varchar(20)", "is_nullable": "NO", "default": "empty_string"},
"allocation_date": {"column_type": "date", "is_nullable": "NO"},
"day_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"overtime_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"night_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"effective_minutes": {"column_type": "decimal(10,2)", "is_nullable": "NO", "default": "zero"},
"good_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"defect_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"scrap_qty": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"changeover_count": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"reference_wage": {"column_type": "decimal(12,2)", "is_nullable": "NO", "default": "zero"},
"created_at": {
"column_type": "datetime",
"is_nullable": "NO",
"default": "current_timestamp",
},
"updated_at": {
"column_type": "datetime",
"is_nullable": "NO",
"default": "current_timestamp",
"extra_contains": ("on update current_timestamp",),
},
}
REQUIRED_INDEXES = {
"idx_report_allocations_report": ("report_id",),
@ -66,11 +79,11 @@ def table_exists(conn) -> bool:
)
def existing_columns(conn) -> set[str]:
def existing_columns(conn) -> dict[str, dict[str, str | None]]:
rows = conn.execute(
text(
"""
SELECT column_name
SELECT column_name, column_type, is_nullable, column_default, extra
FROM information_schema.columns
WHERE table_schema = DATABASE()
AND table_name = :table_name
@ -78,7 +91,15 @@ def existing_columns(conn) -> set[str]:
),
{"table_name": TABLE_NAME},
).all()
return {row[0] for row in rows}
return {
row[0]: {
"column_type": row[1],
"is_nullable": row[2],
"column_default": row[3],
"extra": row[4],
}
for row in rows
}
def existing_indexes(conn) -> dict[str, tuple[str, ...]]:
@ -154,6 +175,25 @@ def create_index(conn, index_name: str, columns: tuple[str, ...]) -> None:
)
def cleanup_duplicate_fk_indexes(conn) -> None:
indexes = existing_indexes(conn)
duplicates = [
index_name
for index_name, columns in indexes.items()
if (
(columns == ("report_id",) and index_name != "idx_report_allocations_report")
or (columns == ("report_item_id",) and index_name != "idx_report_allocations_item")
)
]
for index_name in duplicates:
conn.execute(
text(
f"ALTER TABLE {quote_identifier(TABLE_NAME)} "
f"DROP INDEX {quote_identifier(index_name)}"
)
)
def ensure_schema(conn) -> None:
conn.execute(
text(
@ -195,16 +235,65 @@ def ensure_schema(conn) -> None:
for index_name, columns in REQUIRED_INDEXES.items():
create_index(conn, index_name, columns)
cleanup_duplicate_fk_indexes(conn)
def normalized_default(value: str | None) -> str | None:
if value is None:
return None
lowered = str(value).strip().lower()
if lowered in {"current_timestamp()", "current_timestamp"}:
return "current_timestamp"
try:
return "zero" if float(lowered) == 0 else lowered
except ValueError:
return lowered
def column_matches(column: dict[str, str | None], expected: dict[str, object]) -> list[str]:
mismatches: list[str] = []
actual_type = str(column["column_type"]).lower()
actual_nullable = str(column["is_nullable"]).upper()
actual_default = normalized_default(column["column_default"])
actual_extra = str(column["extra"] or "").lower()
if actual_type != expected["column_type"]:
mismatches.append(f"column_type={column['column_type']}")
if actual_nullable != expected["is_nullable"]:
mismatches.append(f"is_nullable={column['is_nullable']}")
expected_default = expected.get("default")
if expected_default == "empty_string" and actual_default != "":
mismatches.append(f"column_default={column['column_default']!r}")
elif expected_default == "zero" and actual_default != "zero":
mismatches.append(f"column_default={column['column_default']!r}")
elif expected_default == "current_timestamp" and actual_default != "current_timestamp":
mismatches.append(f"column_default={column['column_default']!r}")
for extra_text in expected.get("extra_contains", ()):
if extra_text not in actual_extra:
mismatches.append(f"extra={column['extra']!r}")
return mismatches
def verify_schema(conn) -> None:
if not table_exists(conn):
raise RuntimeError(f"{TABLE_NAME} table is missing")
missing_columns = sorted(REQUIRED_COLUMNS - existing_columns(conn))
columns = existing_columns(conn)
missing_columns = sorted(set(REQUIRED_COLUMNS) - set(columns))
if missing_columns:
raise RuntimeError(f"{TABLE_NAME} missing columns: {', '.join(missing_columns)}")
incompatible_columns = [
f"{column_name}({', '.join(mismatches)})"
for column_name, expected in REQUIRED_COLUMNS.items()
if (mismatches := column_matches(columns[column_name], expected))
]
if incompatible_columns:
raise RuntimeError(f"{TABLE_NAME} incompatible columns: {', '.join(incompatible_columns)}")
indexes = existing_indexes(conn)
missing_indexes = [index_name for index_name in REQUIRED_INDEXES if index_name not in indexes]
if missing_indexes: