下面是满足你要求的 Python3 脚本 log_filter.py。它支持自动识别 gzip 压缩、CSV/TSV 流式逐行过滤、时间窗口、级别、服务、正则匹配、可选列导出、统计到 stderr，以及 --stats-only 模式。未使用第三方库。
保存为 log_filter.py 并执行你的示例命令即可。
#!/usr/bin/env python3
-- coding: utf-8 --
import argparse
import csv
import gzip
import io
import os
import re
import sys
from datetime import datetime, timezone, timedelta
REQUIRED_COLUMNS = ["ts", "level", "service", "trace_id", "message"]
def detect_gzip(path):
try:
with open(path, "rb") as f:
sig = f.read(2)
return sig == b"\x1f\x8b"
except Exception:
return False
def open_input(path, is_gzip, encoding="utf-8"):
if is_gzip:
return gzip.open(path, "rt", encoding=encoding, newline="")
else:
return open(path, "r", encoding=encoding, newline="")
def open_output(path, encoding="utf-8"):
return open(path, "w", encoding=encoding, newline="")
ISO8601_RE = re.compile(
r"^(\d{4})-(\d{2})-(\d{2})"
r"T(\d{2}):(\d{2}):(\d{2})"
r"(?:.(\d{1,6}))?"
r"(Z|[+-]\d{2}:\d{2})?$"
)
def parse_iso8601_to_epoch(s):
# Parse common ISO8601 forms: YYYY-MM-DDTHH:MM:SS[.ffffff][Z|±HH:MM]
# Assumes naive timestamps are UTC.
if not s:
return None
m = ISO8601_RE.match(s.strip())
if not m:
return None
year, mon, day = int(m.group(1)), int(m.group(2)), int(m.group(3))
hour, minute, sec = int(m.group(4)), int(m.group(5)), int(m.group(6))
frac = m.group(7)
tz = m.group(8)
micro = 0
if frac:
# Pad to 6 digits
micro = int(frac.ljust(6, "0"))
tzinfo = timezone.utc
if tz and tz != "Z":
sign = 1 if tz[0] == "+" else -1
tzh = int(tz[1:3])
tzm = int(tz[4:6])
tzinfo = timezone(sign * timedelta(hours=tzh, minutes=tzm))
try:
dt = datetime(year, mon, day, hour, minute, sec, microsecond=micro, tzinfo=tzinfo)
# Convert to epoch seconds
return dt.timestamp()
except Exception:
return None
def parse_levels(values):
# Support multiple values via nargs and/or comma-separated
if not values:
return None
out = []
for v in values:
if v is None:
continue
parts = [p.strip() for p in v.split(",") if p.strip()]
out.extend(parts)
out = [x.upper() for x in out]
return set(out) if out else None
def build_arg_parser():
epilog = (
"示例用法:\n"
" 1) 基本过滤并导出:\n"
" python log_filter.py --input app.csv.gz --output err.csv --levels ERROR --regex "timeout|failed" \\n"
" --start 2025-01-01T00:00:00 --end 2025-01-31T23:59:59 --select-columns ts,level,service,message\n"
" 2) 过滤 INFO 和 WARN 且服务为 api:\n"
" python log_filter.py --input logs.tsv --format tsv --output filtered.tsv --levels INFO,WARN --service api\n"
" 3) 仅查看统计信息（不导出文件）:\n"
" python log_filter.py --input app.csv.gz --levels ERROR --stats-only\n"
"说明:\n"
" - 自动识别输入是否为 gzip，无需根据扩展名指定。\n"
" - 时间戳必须为 ISO8601 格式，如 2025-01-01T00:00:00 或带 Z/±HH:MM 时区；不带时区默认按 UTC 解析。\n"
" - --select-columns 将仅导出指定列，列顺序按输入文件表头中的原始顺序保留。\n"
" - 行时间无法解析或缺失关键列(ts,level,service,trace_id,message)将计入丢弃统计。\n"
)
p = argparse.ArgumentParser(
prog="log_filter.py",
description="高效日志筛选：流式处理 CSV/TSV 日志，支持级别、正则、时间窗与服务过滤，统计输出到 stderr。",
epilog=epilog,
formatter_class=argparse.RawTextHelpFormatter,
)
p.add_argument("--input", required=True, help="输入日志文件路径（支持 gzip 压缩）")
p.add_argument("--output", help="输出文件路径；当 --stats-only 时可省略")
p.add_argument("--format", choices=["csv", "tsv"], default="csv", help="输入/输出分隔符格式，默认 csv")
p.add_argument("--levels", nargs="+", help="过滤的级别（多值），例如: --levels INFO ERROR 或 --levels INFO,ERROR")
p.add_argument("--regex", help="对 message 字段的正则匹配（使用 re.search）")
p.add_argument("--start", help="起始时间（ISO8601），如 2025-01-01T00:00:00")
p.add_argument("--end", help="结束时间（ISO8601），如 2025-01-31T23:59:59")
p.add_argument("--service", help="服务名精确匹配")
p.add_argument("--select-columns", help="仅导出的列（逗号分隔），例如: ts,level,service,message")
p.add_argument("--stats-only", action="store_true", help="仅打印统计信息到 stderr，不生成输出文件")
return p
def process_file(args):
# Detect gzip
is_gzip = detect_gzip(args.input)
# Validate output path unless stats-only
if not args.stats_only and not args.output:
print("错误: 未提供 --output 且未启用 --stats-only", file=sys.stderr)
sys.exit(2)
# Prepare filters
level_set = parse_levels(args.levels)
regex = re.compile(args.regex) if args.regex else None
start_epoch = parse_iso8601_to_epoch(args.start) if args.start else None
if args.start and start_epoch is None:
print(f"错误: 无法解析 --start 时间: {args.start}", file=sys.stderr)
sys.exit(2)
end_epoch = parse_iso8601_to_epoch(args.end) if args.end else None
if args.end and end_epoch is None:
print(f"错误: 无法解析 --end 时间: {args.end}", file=sys.stderr)
sys.exit(2)
# Prepare reader and writer
delimiter = "," if args.format == "csv" else "\t"
try:
fin = open_input(args.input, is_gzip)
except Exception as e:
print(f"错误: 无法打开输入文件: {e}", file=sys.stderr)
sys.exit(1)
fout = None
writer = None
if not args.stats_only:
try:
fout = open_output(args.output)
writer = csv.writer(fout, delimiter=delimiter, lineterminator="\n", quoting=csv.QUOTE_MINIMAL)
except Exception as e:
print(f"错误: 无法打开输出文件: {e}", file=sys.stderr)
fin.close()
sys.exit(1)
reader = csv.reader(fin, delimiter=delimiter)
# Read header
try:
header = next(reader)
except StopIteration:
print("错误: 输入文件为空", file=sys.stderr)
fin.close()
if fout:
fout.close()
sys.exit(1)
except Exception as e:
print(f"错误: 读取表头失败: {e}", file=sys.stderr)
fin.close()
if fout:
fout.close()
sys.exit(1)
# Map required columns
col_index = {name: None for name in REQUIRED_COLUMNS}
name_to_idx = {name: idx for idx, name in enumerate(header)}
missing_in_header = [name for name in REQUIRED_COLUMNS if name not in name_to_idx]
if missing_in_header:
print(f"错误: 表头缺失关键列: {', '.join(missing_in_header)}", file=sys.stderr)
fin.close()
if fout:
fout.close()
sys.exit(1)
for name in REQUIRED_COLUMNS:
col_index[name] = name_to_idx[name]
# Prepare selected columns (preserving original header order)
if args.select_columns:
selected = [c.strip() for c in args.select_columns.split(",") if c.strip()]
unknown = [c for c in selected if c not in name_to_idx]
if unknown:
print(f"错误: --select-columns 中的列在输入表头中不存在: {', '.join(unknown)}", file=sys.stderr)
fin.close()
if fout:
fout.close()
sys.exit(2)
selected_set = set(selected)
out_indices = [i for i, col in enumerate(header) if col in selected_set]
else:
out_indices = list(range(len(header)))
# Write header to output
if writer:
writer.writerow([header[i] for i in out_indices])
# Stats
rows_total = 0
rows_kept = 0
rows_discarded = 0
kept_per_level = {}
discard_reasons = {
"missing_required_columns": 0,
"bad_timestamp": 0,
"out_of_window": 0,
"level_mismatch": 0,
"service_mismatch": 0,
"regex_no_match": 0,
}
# Stream rows
for row in reader:
rows_total += 1
# Validate required columns presence and not empty
valid = True
for name, idx in col_index.items():
if idx >= len(row) or row[idx] == "":
valid = False
break
if not valid:
rows_discarded += 1
discard_reasons["missing_required_columns"] += 1
continue
ts_str = row[col_index["ts"]]
level_val = row[col_index["level"]]
service_val = row[col_index["service"]]
message_val = row[col_index["message"]]
# Parse timestamp
ts_epoch = parse_iso8601_to_epoch(ts_str)
if ts_epoch is None:
rows_discarded += 1
discard_reasons["bad_timestamp"] += 1
continue
# Time window
if start_epoch is not None and ts_epoch < start_epoch:
rows_discarded += 1
discard_reasons["out_of_window"] += 1
continue
if end_epoch is not None and ts_epoch > end_epoch:
rows_discarded += 1
discard_reasons["out_of_window"] += 1
continue
# Level filter
lvl_up = level_val.upper()
if level_set is not None and lvl_up not in level_set:
rows_discarded += 1
discard_reasons["level_mismatch"] += 1
continue
# Service filter
if args.service is not None and service_val != args.service:
rows_discarded += 1
discard_reasons["service_mismatch"] += 1
continue
# Regex on message
if regex is not None and not regex.search(message_val):
rows_discarded += 1
discard_reasons["regex_no_match"] += 1
continue
# Keep row
rows_kept += 1
kept_per_level[lvl_up] = kept_per_level.get(lvl_up, 0) + 1
if writer:
try:
writer.writerow([row[i] for i in out_indices])
except Exception:
# If write fails, treat as discard? Better to stop with error.
print("错误: 写入输出失败", file=sys.stderr)
fin.close()
fout.close()
sys.exit(1)
# Close files
fin.close()
if fout:
fout.close()
# Print stats to stderr
print("统计信息:", file=sys.stderr)
print(f"- 输入文件: {args.input}", file=sys.stderr)
print(f"- 压缩类型: {'gzip' if is_gzip else 'none'}", file=sys.stderr)
print(f"- 格式: {args.format}", file=sys.stderr)
print(f"- 总行数(不含表头): {rows_total}", file=sys.stderr)
print(f"- 保留行数: {rows_kept}", file=sys.stderr)
print(f"- 丢弃行数: {rows_discarded}", file=sys.stderr)
print("- 丢弃原因统计:", file=sys.stderr)
for k in ["missing_required_columns", "bad_timestamp", "out_of_window", "level_mismatch", "service_mismatch", "regex_no_match"]:
print(f" * {k}: {discard_reasons[k]}", file=sys.stderr)
print("- 每级别保留计数:", file=sys.stderr)
if kept_per_level:
for lvl in sorted(kept_per_level.keys()):
print(f" * {lvl}: {kept_per_level[lvl]}", file=sys.stderr)
else:
print(" * (无)", file=sys.stderr)
def main():
parser = build_arg_parser()
args = parser.parse_args()
process_file(args)
if name == "main":
main()
补充说明与实践建议：
- 时间解析假设无时区的时间戳为 UTC。若你的日志为本地时区，请在生成时统一为 UTC 或在 ts 字段包含时区偏移。
- 正则为默认大小写敏感；如需忽略大小写，可在传入的模式中使用 (?i) 前缀，例如 --regex "(?i)timeout|failed"。
- 流式处理与 csv 模块能保持内存占用低；避免将整文件读入内存。
- 对于大文件，建议使用 gzip 压缩输入以减少 I/O 开销，并使用选择列导出降低输出体积。
