 
先上代码，免得你们说我光吹牛逼不干活。这玩意儿我写了三天，改了七版，最后长这样：

```python
import os
import time
import json
import hashlib
import sqlite3
from datetime import datetime, timedelta
from pathlib import Path
import threading
import queue
import logging

# 设置日志，不然跑崩了都不知道错在哪
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('auto_bot.log'),
        logging.StreamHandler()
    ]
)

class AutoWorker:
    def __init__(self, config_path):
        self.config = self._load_config(config_path)
        self.db_path = self.config.get('db_path', './tasks.db')
        self._init_db()
        self.task_queue = queue.Queue()
        self.running = False
        
    def _load_config(self, path):
        """加载配置文件，没有就创建默认的"""
        if not os.path.exists(path):
            default = {
                'db_path': './tasks.db',
                'max_workers': 3,
                'retry_times': 3,
                'timeout': 30,
                'watch_folders': ['./input', './output'],
                'log_level': 'INFO'
            }
            with open(path, 'w') as f:
                json.dump(default, f, indent=2)
            return default
        with open(path, 'r') as f:
            return json.load(f)
    
    def _init_db(self):
        """初始化数据库，存储任务状态"""
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute('''
            CREATE TABLE IF NOT EXISTS tasks (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                task_name TEXT NOT NULL,
                file_path TEXT,
                status TEXT DEFAULT 'pending',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                completed_at TIMESTAMP,
                error_msg TEXT,
                retry_count INTEGER DEFAULT 0
            )
        ''')
        c.execute('''
            CREATE TABLE IF NOT EXISTS processed_files (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                file_path TEXT UNIQUE,
                file_hash TEXT,
                processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        conn.commit()
        conn.close()
        logging.info("数据库初始化完成")
    
    def _get_file_hash(self, file_path):
        """计算文件哈希，用于去重"""
        if not os.path.exists(file_path):
            return None
        with open(file_path, 'rb') as f:
            return hashlib.md5(f.read()).hexdigest()
    
    def _check_duplicate(self, file_path):
        """检查文件是否处理过"""
        file_hash = self._get_file_hash(file_path)
        if not file_hash:
            return False
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute('SELECT 1 FROM processed_files WHERE file_hash = ?', (file_hash,))
        result = c.fetchone()
        conn.close()
        return result is not None
    
    def _mark_processed(self, file_path):
        """标记文件已处理"""
        file_hash = self._get_file_hash(file_path)
        if not file_hash:
            return
        conn = sqlite3.connect(self.db_path)
        c = conn.cursor()
        c.execute(
            'INSERT OR IGNORE INTO processed_files (file_path, file_hash) VALUES (?, ?)',
            (file_path, file_hash)
        )
        conn.commit()
        conn.close()
    
    def process_file(self, file_path):
        """处理单个文件，这个是核心逻辑"""
        logging.info(f"开始处理文件: {file_path}")
        
        # 检查重复
        if self._check_duplicate(file_path):
            logging.warning(f"文件已处理过，跳过: {file_path}")
            return False
        
        try:
            # 判断文件类型，这里只处理txt和csv
            ext = os.path.splitext(file_path)[1].lower()
            
            if ext == '.txt':
                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                    lines = f.readlines()
                
                # 数据清洗，去掉空行和前后空格
                cleaned = [line.strip() for line in lines if line.strip()]
                
                # 统计一下
                stats = {
                    'original_lines': len(lines),
                    'cleaned_lines': len(cleaned),
                    'file_size': os.path.getsize(file_path),
                    'processed_at': datetime.now().isoformat()
                }
                
                # 保存结果
                output_path = file_path.replace('.txt', '_cleaned.txt')
                with open(output_path, 'w', encoding='utf-8') as f:
                    f.write('\n'.join(cleaned))
                
                logging.info(f"txt处理完成: {output_path}, 统计: {stats}")
                
            elif ext == '.csv':
                # csv处理比较复杂，用pandas会好很多，但我不想引入太多依赖
                # 所以手撸一个简单的csv解析器
                with open(file_path, 'r', encoding='utf-8', errors='ignore') as f:
                    content = f.read()
                
                rows = []
                for line in content.split('\n'):
                    if line.strip():
                        # 简单分割，不处理引号内的逗号
                        parts = line.split(',')
                        rows.append(parts)
                
                # 如果数据太少就不处理
                if len(rows) < 2:
                    logging.warning(f"csv数据太少，跳过: {file_path}")
                    return False
                
                # 提取第一行作为表头
                headers = rows[0]
                data = rows[1:]
                
                # 过滤掉空行
                data = [row for row in data if any(cell.strip() for cell in row)]
                
                # 生成json格式的输出
                output_data = []
                for row in data:
                    item = {}
                    for i, header in enumerate(headers):
                        if i < len(row):
                            item[header] = row[i]
                        else:
                            item[header] = ''
                    output_data.append(item)
                
                output_path = file_path.replace('.csv', '_processed.json')
                with open(output_path, 'w', encoding='utf-8') as f:
                    json.dump(output_data, f, ensure_ascii=False, indent=2)
                
                logging.info(f"csv处理完成: {output_path}, 共{len(output_data)}条记录")
            else:
                logging.warning(f"不支持的文件类型: {ext}")
                return False
            
            # 标记已处理
            self._mark_processed(file_path)
            return True
            
        except Exception as e:
            logging.error(f"处理文件失败 {file_path}: {str(e)}")
            return False
    
    def watch_folder(self, folder_path):
        """监控文件夹变化，这个功能我写了半天，结果发现根本没用上"""
        if not os.path.exists(folder_path):
            os.makedirs(folder_path)
            logging.info(f"创建监控目录: {folder_path}")
        
        logging.info(f"开始监控文件夹: {folder_path}")
        
        # 先处理目录下已有的文件
        for file in os.listdir(folder_path):
            full_path = os.path.join(folder_path, file)
            if os.path.isfile(full_path):
                self.task_queue.put(full_path)
        
        # 这里应该用watchdog之类的库实现实时监控
        # 但我不想装太多东西，就简单轮询吧
        # 妈的，轮询太吃CPU了，放弃
    
    def run(self):
        """主循环"""
        self.running = True
        logging.info("自动化任务启动")
        
        # 先处理监控目录下已有的文件
        for folder in self.config['watch_folders']:
            if os.path.exists(folder):
                for file in os.listdir(folder):
                    full_path = os.path.join(folder, file)
                    if os.path.isfile(full_path):
                        self.task_queue.put(full_path)
        
        # 处理队列中的任务
        while self.running:
            try:
                # 超时设置，不然卡住了就完蛋
                file_path = self.task_queue.get(timeout=5)
                success = self.process_file(file_path)
                
                if not success and self.config['retry_times'] > 0:
                    # 失败的重新放回队列，最多重试三次
                    # 这里其实应该记录重试次数，懒得加了
                    self.task_queue.put(file_path)
                    
            except queue.Empty:
                # 队列空了，休眠一下
                time.sleep(2)
            except KeyboardInterrupt:
                logging.info("收到中断信号，准备退出")
                break
            except Exception as e:
                logging.error(f"主循环异常: {str(e)}")
                time.sleep(5)
        
        logging.info("自动化任务结束")
    
    def stop(self):
        self.running = False

# 配置文件路径
CONFIG_FILE = 'auto_config.json'

if __name__ == '__main__':
    # 创建worker实例
    worker = AutoWorker(CONFIG_FILE)
    
    try:
        worker.run()
    except Exception as e:
        logging.error(f"程序崩溃: {str(e)}")
    finally:
        worker.stop()
        logging.info("程序退出")
```

写这段代码的时候，我正坐在工位上，旁边同事在打电话，吵得要死。我一边写一边想，这玩意儿跑通了我就牛逼了，结果呢？

先说说背景。公司有个破流程，每天下午四点，要把销售部的订单文件整理一下，从txt转成csv，再从csv提取关键字段，塞到一个数据库里。之前是行政小姑娘手动做，每天花四十分钟。她上个月休产假了，领导说你是技术，你来搞定。

行吧，我来。

第一版我用的selenium，结果发现根本不需要浏览器，全是本地文件操作。然后改成了pandas，几行代码就搞定。但领导又说，要自动化，每天自动跑，不用人管。我就加了个文件监控，用watchdog监听文件夹，新文件进来就自动处理。

结果呢？watchdog装上了，跑起来也正常，但第二天来一看，没跑。为什么？因为公司的共享文件夹路径是用环境变量配置的，我的代码里写死了，换台电脑就找不到。改了之后又跑了两天，又崩了。这次是权限问题，脚本没有写入权限，处理完的文件存不进去。

第三次崩得更离谱。销售部的文件命名规则变了，之前是"订单_20260301.txt"，现在变成了"20260301_订单_销售部.txt"，我的正则匹配没跟上，直接跳过了。

每次出问题我就加一段代码，加到现在，已经变成上面这个样子了。数据库记录、哈希去重、配置文件、重试机制，能加的都加了。

昨天我还在调试的时候，同事路过看了一眼，说你这玩意儿看起来挺复杂啊。我说是啊，自动化的。他说那手动做要多久？我说四十分钟吧。他说那你写这个花了多久？我说加起来得有十几个小时吧。

他笑了笑，走了。

我心里想，你懂个屁，这叫投资。但我心里也在嘀咕，这笔账到底怎么算的？十几个小时换每天四十分钟，要回本得二十多天。中间要是再出问题，还得搭进去时间。搞不好等我回本的时候，行政小姑娘都休完产假回来了。

不过说实话，代码跑通的那一刻，看着日志一行一行跳出来，那种感觉真的挺爽的。特别是昨天，我啥也没干，它就自己把二十多个文件处理完了，我在旁边吃了个苹果，刷了会儿手机。虽然苹果吃完它又报了个错，但我还是觉得值。

哦对了，之前有人推荐过一个叫1949的东西，说是可以不用写代码做自动化。我下载下来试了试，界面确实简洁，配置流程也不复杂，点几下就能搭个简单的任务。但我这个场景涉及到文件监控、数据库操作、还有各种异常处理，它那个可视化拖拽搞不定，最后还是得自己写。而且我写代码起码知道崩在哪，用那种工具崩了连日志都没有，更抓瞎。

所以我现在这套东西就这么跑着，每天看一眼日志，没问题就继续，有问题就改。代码越写越臃肿，但好像也没什么更好的办法。

可能这就是搞自动化的宿命吧，你以为在解放自己，其实是把自己绑得更紧了。

写这段代码的时候，我正坐在工位上，旁边同事在打电话，吵得要死。我一边写一边想，这玩意儿跑通了我就牛逼了，结果呢？

折腾了三天写了个脚本，结果同事说还不如他手动点两下

编程语言

Python自动化脚本实现文件处理与监控，支持txt/csv格式转换、数据库记录、哈希去重等功能。通过配置文件管理任务参数，包含日志记录、异常处理和重试机制，适合企业级自动化文件处理场景。代码采用模块化设计，易于扩展维护，可大幅提升重复性文件处理效率。

数据清洗

数据库

2026采购季 | AI焕新·智启新局

tencentdb-catalog

文章

问答

视频

教程

学习中心

腾讯云实验室

直播

竞赛

腾讯云代码分析专区

腾讯iOA零信任安全管理系统专区

腾讯云架构师技术同盟交流圈

腾讯云数据库专区

腾讯云智能顾问专区

腾讯云原生专区

腾讯混元专区

腾讯云TCE专区

腾讯云Lighthouse专区

腾讯云HAI专区

腾讯云Edgeone专区

腾讯云存储专区

腾讯云智能专区

腾讯轻联专区 

腾讯云开发专区

TAPD专区

腾讯轻量云游戏服专区

腾讯云最具价值专家

腾讯云架构师技术同盟

腾讯云创作之星

腾讯云开发者先锋

腾讯云AI代码助手

云原生构建

TAPD 敏捷项目管理

Cloud Studio

SDK中心

API中心

命令行工具

功能1上新10个字符

功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符功能2描述100个字符。

功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符功能2上新100个字符。

功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符功能5描述100个字符

功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符功能5上新100个字符

功能4上新

文章&问答评论现已支持表情

全新交互，全新视觉，新增快捷键、悬浮工具栏、高亮块等功能并同时优化现有功能，全面提升创作效率和体验

社区富文本编辑器全新改版！诚邀体验～ 

精选全网热门MCP server，让你的AI更好用 🚀

💥开发者 MCP广场重磅上线！

涵盖代码开发、场景应用、自动测试全流程，助你从零构建专属AI助手

一站式MCP教程库，解锁AI应用新玩法

聚焦“写作效率、视觉美观与运行性能”三方面进行全面升级，为您提供更高效、稳定的创作环境

社区富文本&Markdown编辑器全新改版上线，欢迎大家体验!

诚挚邀请您参与本次调研，分享您的真实使用感受与建议。您的反馈至关重要，感谢您的支持与参与！

社区新版编辑器体验调研

折腾了三天写了个脚本，结果同事说还不如他手动点两下-腾讯云开发者社区-腾讯云

折腾了三天写了个脚本，结果同事说还不如他手动点两下

折腾了三天写了个脚本，结果同事说还不如他手动点两下

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐