Django本地网络应用：CSV文件持续监控入库的后台部署咨询

一、代码放置位置 #

建议将CSV处理逻辑整合到Django的自定义管理命令中，便于结合系统定时任务或监控工具自动触发。具体步骤：

1. 在你的Django应用目录下创建management/commands文件夹（需包含空的__init__.py文件）
2. 新建命令文件（如import_csv.py），将你的CSV处理代码适配后放入该文件

二、持续扫描的实现方案 #

1. 定时扫描（适合非实时需求） #

将CSV导入逻辑封装为Django自定义命令，再通过系统定时任务定期执行，无需手动调用命令行。

适配后的自定义命令代码

# your_app/management/commands/import_csv.py
import os
import shutil
import pandas as pd
from sqlalchemy import create_engine
from django.core.management.base import BaseCommand
from django.conf import settings

class Command(BaseCommand):
    help = '批量导入指定目录下的CSV文件到PostgreSQL'

    def add_arguments(self, parser):
        parser.add_argument('--dirs', nargs='+', default=['C:/', 'D:/', 'E:/'], help='要监控的目标目录列表')

    def handle(self, *args, **options):
        target_dirs = options['dirs']
        for dir_path in target_dirs:
            self.process_single_directory(dir_path)

    def get_csv_files(self, dir_path):
        csv_files = []
        if not os.path.exists(dir_path):
            self.stdout.write(self.style.WARNING(f'目录不存在，跳过: {dir_path}'))
            return csv_files
        for filename in os.listdir(dir_path):
            file_path = os.path.join(dir_path, filename)
            if os.path.isfile(file_path) and filename.endswith('.csv'):
                csv_files.append(file_path)
        return csv_files

    def backup_files(self, csv_files, source_dir):
        backup_root = os.path.join(settings.BASE_DIR, 'BackupFiles')
        backup_dir = os.path.join(backup_root, os.path.basename(source_dir))
        os.makedirs(backup_dir, exist_ok=True)
        
        for file_path in csv_files:
            filename = os.path.basename(file_path)
            dest_path = os.path.join(backup_dir, filename)
            # 处理重名文件
            if os.path.exists(dest_path):
                timestamp = pd.Timestamp.now().strftime('%Y%m%d%H%M%S')
                name, ext = os.path.splitext(filename)
                dest_path = os.path.join(backup_dir, f'{name}_{timestamp}{ext}')
            shutil.move(file_path, dest_path)
        self.stdout.write(self.style.SUCCESS(f'已备份 {len(csv_files)} 个文件到 {backup_dir}'))

    def process_single_directory(self, dir_path):
        csv_files = self.get_csv_files(dir_path)
        if not csv_files:
            self.stdout.write(self.style.INFO(f'{dir_path} 下无CSV文件'))
            return

        # 从Django配置读取数据库信息，避免硬编码
        db_config = settings.DATABASES['default']
        engine = create_engine(
            url=f"postgresql://{db_config['USER']}:{db_config['PASSWORD']}@{db_config['HOST']}:{db_config['PORT']}/{db_config['NAME']}"
        )

        for file_path in csv_files:
            try:
                df = pd.read_csv(file_path, parse_dates=[1, 2], infer_datetime_format=True, encoding="utf-8")
                df.rename(columns={'ItemNumber': 'ItemNumber_id'}, inplace=True)
                df.to_sql('app_logdata', engine, if_exists='append', index=False)
                self.stdout.write(self.style.SUCCESS(f'成功导入文件: {file_path}'))
            except Exception as e:
                self.stdout.write(self.style.ERROR(f'导入失败 {file_path}: {str(e)}'))

        self.backup_files(csv_files, dir_path)

系统定时任务配置

• Windows：打开「任务计划程序」，创建基本任务，触发器设置为「每5分钟/每小时」，操作选择「启动程序」，程序路径填Python解释器，参数填manage.py import_csv --dirs C:/ D:/ E:/，起始目录选你的Django项目根目录。
• Linux：编辑crontab（crontab -e），添加定时规则：

  */5 * * * * cd /path/to/your/django/project && /path/to/python manage.py import_csv --dirs /mnt/c /mnt/d /mnt/e
  

2. 实时监控（适合即时导入需求） #

使用watchdog库监听目录文件变化，一旦有CSV文件创建就自动导入。

实时监控命令代码

# your_app/management/commands/watch_csv.py
import os
import time
import shutil
import pandas as pd
from sqlalchemy import create_engine
from django.core.management.base import BaseCommand
from django.conf import settings
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

class CSVMonitorHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory and event.src_path.endswith('.csv'):
            time.sleep(1)  # 等待文件写入完成
            self.process_csv(event.src_path)

    def process_csv(self, file_path):
        try:
            db_config = settings.DATABASES['default']
            engine = create_engine(
                url=f"postgresql://{db_config['USER']}:{db_config['PASSWORD']}@{db_config['HOST']}:{db_config['PORT']}/{db_config['NAME']}"
            )
            df = pd.read_csv(file_path, parse_dates=[1, 2], infer_datetime_format=True, encoding="utf-8")
            df.rename(columns={'ItemNumber': 'ItemNumber_id'}, inplace=True)
            df.to_sql('app_logdata', engine, if_exists='append', index=False)
            print(f'成功导入: {file_path}')

            # 备份文件
            source_dir = os.path.dirname(file_path)
            backup_dir = os.path.join(settings.BASE_DIR, 'BackupFiles', os.path.basename(source_dir))
            os.makedirs(backup_dir, exist_ok=True)
            filename = os.path.basename(file_path)
            dest_path = os.path.join(backup_dir, filename)
            if os.path.exists(dest_path):
                timestamp = pd.Timestamp.now().strftime('%Y%m%d%H%M%S')
                name, ext = os.path.splitext(filename)
                dest_path = os.path.join(backup_dir, f'{name}_{timestamp}{ext}')
            shutil.move(file_path, dest_path)
            print(f'已备份到: {dest_path}')
        except Exception as e:
            print(f'处理失败 {file_path}: {str(e)}')

class Command(BaseCommand):
    help = '实时监控指定目录下的CSV文件并自动导入'

    def add_arguments(self, parser):
        parser.add_argument('--dirs', nargs='+', default=['C:/', 'D:/', 'E:/'], help='要监控的目录列表')

    def handle(self, *args, **options):
        target_dirs = options['dirs']
        event_handler = CSVMonitorHandler()
        observer = Observer()

        for dir_path in target_dirs:
            if os.path.exists(dir_path):
                observer.schedule(event_handler, dir_path, recursive=False)
                self.stdout.write(self.style.SUCCESS(f'开始监控: {dir_path}'))
            else:
                self.stdout.write(self.style.WARNING(f'目录不存在，跳过: {dir_path}'))

        observer.start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

后台运行配置

• Windows：用pywin32将脚本注册为Windows服务，实现开机自启。
• Linux：用systemd创建守护进程，配置自动重启。

三、Django REST Framework (DRF) 适用性分析 #

DRF的核心定位是构建RESTful API接口，你的场景核心需求是本地文件监控+数据导入，如果不需要对外提供HTTP接口（比如手动触发导入、查询数据），则完全不需要使用DRF。

若后续需要通过API实现以下功能，可考虑引入DRF：

• 手动触发CSV导入任务
• 查询已导入的数据记录
• 获取导入日志

四、现有代码优化点 #

1. 数据库配置解耦：用settings.DATABASES读取数据库信息，避免硬编码账号密码。
2. 多目录支持：原代码仅扫描当前工作目录，改为接收指定目录参数，适配C:/、D:/等多盘符需求。
3. 异常处理：添加try-except捕获文件读取、数据库导入的异常，避免任务崩溃。
4. 重名文件处理：备份时对重名文件添加时间戳，防止覆盖。
5. 跨平台路径：用os.path.join拼接路径，兼容Windows和Linux的路径分隔符。

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