from pathlib import Path import sys from sqlalchemy import text ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from app.database import engine # noqa: E402 from app.services.attendance_points import DEFAULT_ATTENDANCE_POINT_NAME # noqa: E402 TARGET_COLLATION = "utf8mb4_unicode_ci" def column_exists(conn, table: str, column: str) -> bool: return bool(conn.execute(text(""" SELECT COUNT(*) FROM information_schema.columns WHERE table_schema = DATABASE() AND table_name = :table AND column_name = :column """), {"table": table, "column": column}).scalar_one()) def table_exists(conn, table: str) -> bool: return bool(conn.execute(text(""" SELECT COUNT(*) FROM information_schema.tables WHERE table_schema = DATABASE() AND table_name = :table """), {"table": table}).scalar_one()) def primary_columns(conn, table: str) -> list[str]: rows = conn.execute(text(""" SELECT column_name FROM information_schema.key_column_usage WHERE table_schema = DATABASE() AND table_name = :table AND constraint_name = 'PRIMARY' ORDER BY ordinal_position """), {"table": table}).all() return [row[0] for row in rows] def index_exists(conn, table: str, index_name: str) -> bool: return bool(conn.execute(text(""" SELECT COUNT(*) FROM information_schema.statistics WHERE table_schema = DATABASE() AND table_name = :table AND index_name = :index_name """), {"table": table, "index_name": index_name}).scalar_one()) def create_index(conn, table: str, index_name: str, columns: str) -> None: if not index_exists(conn, table, index_name): conn.execute(text(f"CREATE INDEX {index_name} ON {table} ({columns})")) def add_point_column(conn, table: str, after: str | None = None) -> None: if column_exists(conn, table, "attendance_point_name"): return after_clause = f" AFTER {after}" if after else "" conn.execute(text( f"ALTER TABLE {table} " f"ADD COLUMN attendance_point_name VARCHAR(128) COLLATE {TARGET_COLLATION} NOT NULL DEFAULT :default_name{after_clause}" ), {"default_name": DEFAULT_ATTENDANCE_POINT_NAME}) def ensure_primary(conn, table: str, columns: list[str]) -> None: if primary_columns(conn, table) == columns: return if primary_columns(conn, table): conn.execute(text(f"ALTER TABLE {table} DROP PRIMARY KEY")) joined = ", ".join(columns) conn.execute(text(f"ALTER TABLE {table} ADD PRIMARY KEY ({joined})")) def main() -> None: with engine.begin() as conn: conn.execute(text(""" CREATE TABLE IF NOT EXISTS attendance_points ( name VARCHAR(128) NOT NULL PRIMARY KEY, latitude DECIMAL(10, 7) NULL, longitude DECIMAL(10, 7) NULL, radius_meters INT NOT NULL DEFAULT 500, day_start VARCHAR(5) NOT NULL DEFAULT '08:00', day_end VARCHAR(5) NOT NULL DEFAULT '17:20', lunch_start VARCHAR(5) NOT NULL DEFAULT '11:40', lunch_end VARCHAR(5) NOT NULL DEFAULT '12:40', dinner_start VARCHAR(5) NOT NULL DEFAULT '17:20', dinner_end VARCHAR(5) NOT NULL DEFAULT '18:00', overtime_start VARCHAR(5) NOT NULL DEFAULT '18:00', overtime_end VARCHAR(5) NOT NULL DEFAULT '20:00', night_start VARCHAR(5) NOT NULL DEFAULT '20:00', night_end VARCHAR(5) NOT NULL DEFAULT '06:00', remark VARCHAR(500) NULL, is_active BOOLEAN NOT NULL DEFAULT TRUE, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """)) conn.execute(text(""" INSERT IGNORE INTO attendance_points ( name, latitude, longitude, radius_meters, remark, is_active ) SELECT :name, attendance_latitude, attendance_longitude, COALESCE(attendance_radius_meters, 500), '默认考勤点', TRUE FROM work_schedules WHERE id = 1 """), {"name": DEFAULT_ATTENDANCE_POINT_NAME}) conn.execute(text(""" INSERT IGNORE INTO attendance_points (name, radius_meters, remark, is_active) VALUES (:name, 500, '默认考勤点', TRUE) """), {"name": DEFAULT_ATTENDANCE_POINT_NAME}) conn.execute(text(""" CREATE TABLE IF NOT EXISTS person_attendance_points ( phone VARCHAR(20) NOT NULL, attendance_point_name VARCHAR(128) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (phone, attendance_point_name), INDEX idx_person_points_point (attendance_point_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """)) conn.execute(text(""" INSERT IGNORE INTO person_attendance_points (phone, attendance_point_name) SELECT phone, :name FROM personnel """), {"name": DEFAULT_ATTENDANCE_POINT_NAME}) if table_exists(conn, "products"): add_point_column(conn, "products") ensure_primary(conn, "products", [ "attendance_point_name", "project_no", "product_name", "device_no", "process_name", ]) create_index(conn, "products", "idx_products_point", "attendance_point_name") if table_exists(conn, "equipment"): add_point_column(conn, "equipment") ensure_primary(conn, "equipment", ["attendance_point_name", "device_no"]) create_index(conn, "equipment", "idx_equipment_point", "attendance_point_name") if table_exists(conn, "work_sessions"): add_point_column(conn, "work_sessions", "id") create_index(conn, "work_sessions", "idx_sessions_point_status", "attendance_point_name, status") if table_exists(conn, "work_session_devices"): add_point_column(conn, "work_session_devices", "session_id") create_index(conn, "work_session_devices", "idx_session_devices_point", "attendance_point_name") if table_exists(conn, "production_reports"): add_point_column(conn, "production_reports", "id") create_index(conn, "production_reports", "idx_reports_point_status", "attendance_point_name, status, report_date") if table_exists(conn, "production_report_items"): add_point_column(conn, "production_report_items", "report_id") conn.execute(text(""" UPDATE production_report_items item JOIN production_reports report ON report.id = item.report_id SET item.attendance_point_name = report.attendance_point_name WHERE item.attendance_point_name = '' OR item.attendance_point_name IS NULL """)) create_index(conn, "production_report_items", "idx_report_items_point_product", "attendance_point_name, project_no, product_name") if table_exists(conn, "device_qrcodes"): add_point_column(conn, "device_qrcodes") ensure_primary(conn, "device_qrcodes", ["attendance_point_name", "device_no", "process_name"]) conn.execute(text(""" CREATE TABLE IF NOT EXISTS notice_points ( notice_id BIGINT NOT NULL, attendance_point_name VARCHAR(128) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, PRIMARY KEY (notice_id, attendance_point_name), INDEX idx_notice_points_point (attendance_point_name) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci """)) conn.execute(text(""" INSERT IGNORE INTO notice_points (notice_id, attendance_point_name) SELECT id, :name FROM notices """), {"name": DEFAULT_ATTENDANCE_POINT_NAME}) if table_exists(conn, "reconciliation_ledger_entries"): add_point_column(conn, "reconciliation_ledger_entries") ensure_primary(conn, "reconciliation_ledger_entries", [ "attendance_point_name", "year", "month", "product_name", ]) print("attendance points migrated") if __name__ == "__main__": main()