Python脚本自动按日期打开对应.txt并同步数据至SQL Server

1. 自动生成当日目标文件路径 #

利用Python的datetime模块动态拼接当日文件名，彻底告别手动修改路径：

from datetime import datetime

# 替换为你的CNC文件实际存储目录
FILE_STORAGE_DIR = "/mnt/cnc_data/logs"

def get_today_file_path():
    today_date_str = datetime.now().strftime("%Y-%m-%d")
    return f"{FILE_STORAGE_DIR}/industry_{today_date_str}.txt"

2. 高效读取文件最后一行 #

针对CNC日志可能较大的情况，用文件指针定位的方式直接读取最后一行，避免加载整个文件：

def get_last_line(file_path):
    try:
        with open(file_path, "r", encoding="utf-8") as f:
            # 定位到文件末尾
            f.seek(0, 2)
            pos = f.tell()
            # 往前查找换行符，确定最后一行起始位置
            while pos > 0:
                pos -= 1
                f.seek(pos)
                if f.read(1) == "\n":
                    break
            last_line = f.read().strip()
            return last_line if last_line else None
    except FileNotFoundError:
        print(f"提示：当日日志文件 {file_path} 尚未生成")
        return None

3. 数据写入SQL Server #

使用pyodbc库连接数据库，先通过pip install pyodbc安装依赖，代码示例如下：

import pyodbc

# 替换为你的数据库实际配置
DB_SETTINGS = {
    "server": "CNC-DB-SERVER",
    "db_name": "CNC_Production_DB",
    "user": "db_user",
    "pwd": "db_password",
    "driver": "{ODBC Driver 17 for SQL Server}"
}

def save_to_sql(data):
    if not data:
        return
    conn_str = f"DRIVER={DB_SETTINGS['driver']};SERVER={DB_SETTINGS['server']};DATABASE={DB_SETTINGS['db_name']};UID={DB_SETTINGS['user']};PWD={DB_SETTINGS['pwd']}"
    try:
        with pyodbc.connect(conn_str) as conn:
            cursor = conn.cursor()
            # 替换为你的实际表结构和字段名
            insert_sql = "INSERT INTO cnc_real_time_data (data_content, collect_time) VALUES (?, GETDATE())"
            cursor.execute(insert_sql, (data,))
            conn.commit()
            print(f"数据已写入数据库：{data}")
    except Exception as e:
        print(f"数据库写入失败：{str(e)}")

4. 定时任务与后台运行 #

用循环+休眠实现每5秒执行一次，跨天场景下会自动切换当日文件：

import time

def main():
    while True:
        today_file = get_today_file_path()
        latest_data = get_last_line(today_file)
        save_to_sql(latest_data)
        time.sleep(5)

if __name__ == "__main__":
    main()

额外优化建议 #

• 后台运行：Linux下用nohup python3 your_script.py &；Windows下可封装成系统服务，避免窗口关闭后脚本停止。
• 日志记录：引入logging模块，将运行日志写入文件，方便后续排查问题。
• 异常重试：针对数据库连接失败的情况，添加重试逻辑，提升稳定性。

内容的提问来源于stack exchange，提问作者Campenhe