本文目录导读:

我需要更具体的需求才能给出准确方案,请补充以下信息:
日志来源是什么?
- 数据库(MySQL、PostgreSQL、Elasticsearch、MongoDB...)
- 搜索引擎(Elasticsearch、Splunk、Solr...)
- Web服务器(Nginx、Apache、IIS...)
- 系统命令(
grep、journalctl、find...) - 云平台(AWS CloudWatch、阿里云SLS...)
检索条件是什么?
- 按时间范围(如
2024-01-01 至 2024-01-31) - 按关键字(如
error、timeout) - 按级别(如
ERROR、WARN、INFO) - 按来源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")