脚本怎样批量导出检索日志

wen 实用脚本 30

本文目录导读:

脚本怎样批量导出检索日志

  1. 先给一个通用示例(Python + 任意文本日志)
  2. 如果是 Elasticsearch(ES)

我需要更具体的需求才能给出准确方案,请补充以下信息:

日志来源是什么?

  • 数据库(MySQL、PostgreSQL、Elasticsearch、MongoDB...)
  • 搜索引擎(Elasticsearch、Splunk、Solr...)
  • Web服务器(Nginx、Apache、IIS...)
  • 系统命令(grepjournalctlfind...)
  • 云平台(AWS CloudWatch、阿里云SLS...)

检索条件是什么?

  • 按时间范围(如 2024-01-01 至 2024-01-31
  • 按关键字(如 errortimeout
  • 按级别(如 ERRORWARNINFO
  • 按来源IP、用户名、请求ID等

导出格式要求?

  • 文本文件(.txt, .log)
  • CSV / JSON
  • Excel
  • 直接压缩打包

运行环境?

  • Windows(PowerShell、Batch)
  • Linux/macOS(Bash、Python)
  • 跨平台(Python、Go、Node.js)

先给一个通用示例(Python + 任意文本日志)

假设日志是纯文本文件,按日期分割存储(如 app-2024-01-01.log),需要批量导出包含 ERROR 的行:

import os
import shutil
from datetime import datetime, timedelta
# 配置
log_dir = "/var/log/myapp"
output_dir = "/tmp/exported_logs"
keyword = "ERROR"
date_start = "2024-01-01"
date_end = "2024-01-31"
os.makedirs(output_dir, exist_ok=True)
start = datetime.strptime(date_start, "%Y-%m-%d")
end = datetime.strptime(date_end, "%Y-%m-%d")
current = start
while current <= end:
    filename = f"app-{current.strftime('%Y-%m-%d')}.log"
    filepath = os.path.join(log_dir, filename)
    if os.path.exists(filepath):
        output_file = os.path.join(output_dir, f"filtered-{filename}")
        with open(filepath, 'r', encoding='utf-8') as inf, \
             open(output_file, 'w', encoding='utf-8') as outf:
            for line in inf:
                if keyword in line:
                    outf.write(line)
        print(f"[OK] {filename} -> {output_file}")
    else:
        print(f"[SKIP] {filename} not found")
    current += timedelta(days=1)

如果是 Elasticsearch(ES)

使用 elasticsearch-dump 或 Python elasticsearch 库:

from elasticsearch import Elasticsearch
import json
es = Elasticsearch(["http://localhost:9200"])
index_pattern = "logs-*"
query = {
    "query": {"match": {"message": "ERROR"}},
    "size": 10000
}
result = es.search(index=index_pattern, body=query, scroll="2m")
with open("export.json", "w") as f:
    for hit in result['hits']['hits']:
        f.write(json.dumps(hit['_source']) + "\n")

抱歉,评论功能暂时关闭!