꼬리와 비슷한 파이썬으로 파일의 마지막 n 줄을 얻습니다.
https://stackoverflow.com/questions/136168
italiano
english
français
española
中国
日本の
العربية
Deutsch
한국어
Português
Russian문제
웹 응용 프로그램을 위해 로그 파일 뷰어를 작성하고 로그 파일의 줄을 통해 페이지를 찍고 싶습니다. 파일의 항목은 하단의 최신 항목을 기반으로 한 줄입니다.
그래서 나는 필요하다 tail() 읽을 수있는 방법 n 하단에서 라인과 오프셋을 지원합니다. 내가 생각한 것은 다음과 같이 보입니다.
def tail(f, n, offset=0):
"""Reads a n lines from f with an offset of offset lines."""
avg_line_length = 74
to_read = n + offset
while 1:
try:
f.seek(-(avg_line_length * to_read), 2)
except IOError:
# woops. apparently file is smaller than what we want
# to step back, go to the beginning instead
f.seek(0)
pos = f.tell()
lines = f.read().splitlines()
if len(lines) >= to_read or pos == 0:
return lines[-to_read:offset and -offset or None]
avg_line_length *= 1.3
이것이 합리적인 접근 방식입니까? 오프셋으로 파일을 꼬리 로그 로그를 맞추는 권장 방법은 무엇입니까?
해결책 6
내가 사용한 코드. 나는 이것이 지금까지 최고라고 생각합니다.
def tail(f, n, offset=None):
"""Reads a n lines from f with an offset of offset lines. The return
value is a tuple in the form ``(lines, has_more)`` where `has_more` is
an indicator that is `True` if there are more lines in the file.
"""
avg_line_length = 74
to_read = n + (offset or 0)
while 1:
try:
f.seek(-(avg_line_length * to_read), 2)
except IOError:
# woops. apparently file is smaller than what we want
# to step back, go to the beginning instead
f.seek(0)
pos = f.tell()
lines = f.read().splitlines()
if len(lines) >= to_read or pos == 0:
return lines[-to_read:offset and -offset or None], \
len(lines) > to_read or pos > 0
avg_line_length *= 1.3
다른 팁
이것은 당신보다 더 빠를 수 있습니다. 라인 길이에 대한 가정이 없습니다. ' n'문자의 올바른 수를 찾을 때까지 파일 한 번에 한 블록 씩 후면.
def tail( f, lines=20 ):
total_lines_wanted = lines
BLOCK_SIZE = 1024
f.seek(0, 2)
block_end_byte = f.tell()
lines_to_go = total_lines_wanted
block_number = -1
blocks = [] # blocks of size BLOCK_SIZE, in reverse order starting
# from the end of the file
while lines_to_go > 0 and block_end_byte > 0:
if (block_end_byte - BLOCK_SIZE > 0):
# read the last block we haven't yet read
f.seek(block_number*BLOCK_SIZE, 2)
blocks.append(f.read(BLOCK_SIZE))
else:
# file too small, start from begining
f.seek(0,0)
# only read what was not read
blocks.append(f.read(block_end_byte))
lines_found = blocks[-1].count('\n')
lines_to_go -= lines_found
block_end_byte -= BLOCK_SIZE
block_number -= 1
all_read_text = ''.join(reversed(blocks))
return '\n'.join(all_read_text.splitlines()[-total_lines_wanted:])
나는 실질적인 문제로 당신이 그런 것을 알 수 없을 때 선 길이에 대한 까다로운 가정을 좋아하지 않습니다.
일반적으로 이것은 루프를 통해 첫 번째 또는 두 번째 패스에서 마지막 20 줄을 찾습니다. 74 캐릭터가 실제로 정확하다면 블록 크기 2048을 만들고 거의 즉시 20 줄을 꼬리게 할 것입니다.
또한, 나는 물리적 OS 블록과 정렬을 정렬하려고 많은 뇌 칼로리를 태우지 않습니다. 이러한 높은 수준의 I/O 패키지를 사용하면 OS 블록 경계에 맞추려고 시도한 성능 결과가 보일 것입니다. 하위 레벨 I/O를 사용하면 속도가 빨라질 수 있습니다.
Python 2의 Unix와 같은 시스템을 가정합니다.
import os
def tail(f, n, offset=0):
stdin,stdout = os.popen2("tail -n "+n+offset+" "+f)
stdin.close()
lines = stdout.readlines(); stdout.close()
return lines[:,-offset]
Python 3의 경우 다음에 할 수 있습니다.
import subprocess
def tail(f, n, offset=0):
proc = subprocess.Popen(['tail', '-n', n + offset, f], stdout=subprocess.PIPE)
lines = proc.stdout.readlines()
return lines[:, -offset]
전체 파일을 읽을 수있는 경우 Deque를 사용하십시오.
from collections import deque
deque(f, maxlen=n)
2.6 이전에 Deques는 Maxlen 옵션이 없었지만 구현하기에 쉽습니다.
import itertools
def maxque(items, size):
items = iter(items)
q = deque(itertools.islice(items, size))
for item in items:
del q[0]
q.append(item)
return q
끝에서 파일을 읽어야하는 경우 갤럽 (일명 지수) 검색을 사용하십시오.
def tail(f, n):
assert n >= 0
pos, lines = n+1, []
while len(lines) <= n:
try:
f.seek(-pos, 2)
except IOError:
f.seek(0)
break
finally:
lines = list(f)
pos *= 2
return lines[-n:]
여기 내 대답이 있습니다. 순수한 파이썬. Timeit을 사용하면 꽤 빠릅니다. 100,000 줄이있는 로그 파일의 100 줄을 테일링합니다.
>>> timeit.timeit('tail.tail(f, 100, 4098)', 'import tail; f = open("log.txt", "r");', number=10)
0.0014600753784179688
>>> timeit.timeit('tail.tail(f, 100, 4098)', 'import tail; f = open("log.txt", "r");', number=100)
0.00899195671081543
>>> timeit.timeit('tail.tail(f, 100, 4098)', 'import tail; f = open("log.txt", "r");', number=1000)
0.05842900276184082
>>> timeit.timeit('tail.tail(f, 100, 4098)', 'import tail; f = open("log.txt", "r");', number=10000)
0.5394978523254395
>>> timeit.timeit('tail.tail(f, 100, 4098)', 'import tail; f = open("log.txt", "r");', number=100000)
5.377126932144165
코드는 다음과 같습니다.
import os
def tail(f, lines=1, _buffer=4098):
"""Tail a file and get X lines from the end"""
# place holder for the lines found
lines_found = []
# block counter will be multiplied by buffer
# to get the block size from the end
block_counter = -1
# loop until we find X lines
while len(lines_found) < lines:
try:
f.seek(block_counter * _buffer, os.SEEK_END)
except IOError: # either file is too small, or too many lines requested
f.seek(0)
lines_found = f.readlines()
break
lines_found = f.readlines()
# we found enough lines, get out
# Removed this line because it was redundant the while will catch
# it, I left it for history
# if len(lines_found) > lines:
# break
# decrement the block counter to get the
# next X bytes
block_counter -= 1
return lines_found[-lines:]
위의 S.Lott의 대답은 거의 나에게 효과가 있지만 결국 부분 선을 준다. 데이터는 읽기 블록을 역 순서로 유지하기 때문에 블록 경계에 대한 데이터를 손상 시킨다는 것이 밝혀졌습니다. ''.join (데이터)이 호출되면 블록은 잘못된 순서입니다. 이것은 그것을 해결합니다.
def tail(f, window=20):
"""
Returns the last `window` lines of file `f` as a list.
f - a byte file-like object
"""
if window == 0:
return []
BUFSIZ = 1024
f.seek(0, 2)
bytes = f.tell()
size = window + 1
block = -1
data = []
while size > 0 and bytes > 0:
if bytes - BUFSIZ > 0:
# Seek back one whole BUFSIZ
f.seek(block * BUFSIZ, 2)
# read BUFFER
data.insert(0, f.read(BUFSIZ))
else:
# file too small, start from begining
f.seek(0,0)
# only read what was not read
data.insert(0, f.read(bytes))
linesFound = data[0].count('\n')
size -= linesFound
bytes -= BUFSIZ
block -= 1
return ''.join(data).splitlines()[-window:]
MMAP를 사용한 간단하고 빠른 솔루션 :
import mmap
import os
def tail(filename, n):
"""Returns last n lines from the filename. No exception handling"""
size = os.path.getsize(filename)
with open(filename, "rb") as f:
# for Windows the mmap parameters are different
fm = mmap.mmap(f.fileno(), 0, mmap.MAP_SHARED, mmap.PROT_READ)
try:
for i in xrange(size - 1, -1, -1):
if fm[i] == '\n':
n -= 1
if n == -1:
break
return fm[i + 1 if i else 0:].splitlines()
finally:
fm.close()
삽입 및 추가 및 반전되는 더 깨끗한 Python3 호환 버전 :
def tail(f, window=1):
"""
Returns the last `window` lines of file `f` as a list of bytes.
"""
if window == 0:
return b''
BUFSIZE = 1024
f.seek(0, 2)
end = f.tell()
nlines = window + 1
data = []
while nlines > 0 and end > 0:
i = max(0, end - BUFSIZE)
nread = min(end, BUFSIZE)
f.seek(i)
chunk = f.read(nread)
data.append(chunk)
nlines -= chunk.count(b'\n')
end -= nread
return b'\n'.join(b''.join(reversed(data)).splitlines()[-window:])
다음과 같이 사용하십시오.
with open(path, 'rb') as f:
last_lines = tail(f, 3).decode('utf-8')
위의 Popen이 최상의 솔루션이라는 것을 알았습니다. 빠르고 더럽고 유닉스 머신에서 Python 2.6에서 작동합니다. 다음을 사용했습니다.
def GetLastNLines(self, n, fileName):
"""
Name: Get LastNLines
Description: Gets last n lines using Unix tail
Output: returns last n lines of a file
Keyword argument:
n -- number of last lines to return
filename -- Name of the file you need to tail into
"""
p=subprocess.Popen(['tail','-n',str(n),self.__fileName], stdout=subprocess.PIPE)
soutput,sinput=p.communicate()
return soutput
Soutput은 코드의 마지막 n 줄을 포함합니다. Soutput 라인을 통해 라인별로 반복하기 위해 :
for line in GetLastNLines(50,'myfile.log').split('\n'):
print line
Python3에 대한 @papercrane 솔루션을 업데이트하십시오. 파일을 엽니 다 open(filename, 'rb') 그리고:
def tail(f, window=20):
"""Returns the last `window` lines of file `f` as a list.
"""
if window == 0:
return []
BUFSIZ = 1024
f.seek(0, 2)
remaining_bytes = f.tell()
size = window + 1
block = -1
data = []
while size > 0 and remaining_bytes > 0:
if remaining_bytes - BUFSIZ > 0:
# Seek back one whole BUFSIZ
f.seek(block * BUFSIZ, 2)
# read BUFFER
bunch = f.read(BUFSIZ)
else:
# file too small, start from beginning
f.seek(0, 0)
# only read what was not read
bunch = f.read(remaining_bytes)
bunch = bunch.decode('utf-8')
data.insert(0, bunch)
size -= bunch.count('\n')
remaining_bytes -= BUFSIZ
block -= 1
return ''.join(data).splitlines()[-window:]
주석가의 명령에 따라 답변을 게시합니다 비슷한 질문에 대한 나의 대답 동일한 기술이 사용되어 파일의 마지막 줄을 돌연변이하는 것이 아니라 단순히 얻는 것이 아닙니다.
크기가 상당한 파일의 경우 mmap 이 작업을 수행하는 가장 좋은 방법입니다. 기존의 개선 mmap 답변,이 버전은 Windows와 Linux 사이에서 휴대 할 수 있으며 더 빠르게 실행해야합니다 (GB 범위의 파일이있는 32 비트 파이썬에서 일부 수정 없이는 작동하지 않지만 이 처리에 대한 힌트에 대한 다른 답변 및 Python 2에서 작업하기 위해 수정하십시오.).
import io # Gets consistent version of open for both Py2.7 and Py3.x
import itertools
import mmap
def skip_back_lines(mm, numlines, startidx):
'''Factored out to simplify handling of n and offset'''
for _ in itertools.repeat(None, numlines):
startidx = mm.rfind(b'\n', 0, startidx)
if startidx < 0:
break
return startidx
def tail(f, n, offset=0):
# Reopen file in binary mode
with io.open(f.name, 'rb') as binf, mmap.mmap(binf.fileno(), 0, access=mmap.ACCESS_READ) as mm:
# len(mm) - 1 handles files ending w/newline by getting the prior line
startofline = skip_back_lines(mm, offset, len(mm) - 1)
if startofline < 0:
return [] # Offset lines consumed whole file, nothing to return
# If using a generator function (yield-ing, see below),
# this should be a plain return, no empty list
endoflines = startofline + 1 # Slice end to omit offset lines
# Find start of lines to capture (add 1 to move from newline to beginning of following line)
startofline = skip_back_lines(mm, n, startofline) + 1
# Passing True to splitlines makes it return the list of lines without
# removing the trailing newline (if any), so list mimics f.readlines()
return mm[startofline:endoflines].splitlines(True)
# If Windows style \r\n newlines need to be normalized to \n, and input
# is ASCII compatible, can normalize newlines with:
# return mm[startofline:endoflines].replace(os.linesep.encode('ascii'), b'\n').splitlines(True)
이것은 꼬리가 작은 줄의 수가 충분히 작다고 가정합니다. 한 번에 모두 메모리에 안전하게 읽을 수 있습니다. 또한 발전기 기능을 만들고 최종 라인을 대체하여 한 번에 한 줄을 수동으로 읽을 수 있습니다.
mm.seek(startofline)
# Call mm.readline n times, or until EOF, whichever comes first
# Python 3.2 and earlier:
for line in itertools.islice(iter(mm.readline, b''), n):
yield line
# 3.3+:
yield from itertools.islice(iter(mm.readline, b''), n)
마지막으로, 이것은 이진 모드로 읽습니다 (사용하는 데 필요합니다. mmap) 그래서 그것은 제공합니다 str 라인 (py2) 및 bytes 라인 (py3); 네가 원한다면 unicode (py2) 또는 str (py3), 반복적 인 접근법은 당신을 위해 해독하고/또는 newlines를 고정하기 위해 조정할 수 있습니다.
lines = itertools.islice(iter(mm.readline, b''), n)
if f.encoding: # Decode if the passed file was opened with a specific encoding
lines = (line.decode(f.encoding) for line in lines)
if 'b' not in f.mode: # Fix line breaks if passed file opened in text mode
lines = (line.replace(os.linesep, '\n') for line in lines)
# Python 3.2 and earlier:
for line in lines:
yield line
# 3.3+:
yield from lines
참고 : 테스트 할 파이썬에 액세스 할 수없는 기계 에이 모든 것을 입력했습니다. 내가 뭐든지 유인했다면 알려주세요. 이것은 충분히 비슷했습니다 내 다른 대답 그게 나 생각한다 작동해야하지만 조정 (예 : 처리 offset) 미묘한 오류로 이어질 수 있습니다. 실수가 있으면 의견에 알려주십시오.
S.Lott의 최고 투표 답변 (21:43 년 9 월 25 일)을 기반으로하지만 작은 파일에 대해 수정되었습니다.
def tail(the_file, lines_2find=20):
the_file.seek(0, 2) #go to end of file
bytes_in_file = the_file.tell()
lines_found, total_bytes_scanned = 0, 0
while lines_2find+1 > lines_found and bytes_in_file > total_bytes_scanned:
byte_block = min(1024, bytes_in_file-total_bytes_scanned)
the_file.seek(-(byte_block+total_bytes_scanned), 2)
total_bytes_scanned += byte_block
lines_found += the_file.read(1024).count('\n')
the_file.seek(-total_bytes_scanned, 2)
line_list = list(the_file.readlines())
return line_list[-lines_2find:]
#we read at least 21 line breaks from the bottom, block by block for speed
#21 to ensure we don't get a half line
이것이 유용하기를 바랍니다.
PIP를 사용하여 설치할 수있는 PYPI에는 기존의 꼬리 구현이 있습니다.
- mtfileutil
- 멀티 테일
- log4 tailer
- ...
상황에 따라 기존 도구 중 하나를 사용하면 이점이있을 수 있습니다.
다음은 매우 간단한 구현입니다.
with open('/etc/passwd', 'r') as f:
try:
f.seek(0,2)
s = ''
while s.count('\n') < 11:
cur = f.tell()
f.seek((cur - 10))
s = f.read(10) + s
f.seek((cur - 10))
print s
except Exception as e:
f.readlines()
단순한 :
with open("test.txt") as f:
data = f.readlines()
tail = data[-2:]
print(''.join(tail)
매우 큰 파일의 효율성 (꼬리를 사용하려는 로그 파일 상황에서 일반적으로)의 경우 일반적으로 전체 파일을 읽지 않으려 고합니다 (전체 파일을 한 번에 메모리에 읽지 않고도 수행하더라도). 어떻게 든 캐릭터가 아닌 라인으로 오프셋을 해결해야합니다. 한 가지 가능성은 Char의 Seek () char와 함께 거꾸로 읽는 것이지만 이것은 매우 느립니다. 대신, 더 큰 블록으로 처리하는 것이 좋습니다.
나는 여기에서 사용할 수있는 파일을 뒤로 읽기 위해 얼마 전에 쓴 유틸리티 기능을 사용합니다.
import os, itertools
def rblocks(f, blocksize=4096):
"""Read file as series of blocks from end of file to start.
The data itself is in normal order, only the order of the blocks is reversed.
ie. "hello world" -> ["ld","wor", "lo ", "hel"]
Note that the file must be opened in binary mode.
"""
if 'b' not in f.mode.lower():
raise Exception("File must be opened using binary mode.")
size = os.stat(f.name).st_size
fullblocks, lastblock = divmod(size, blocksize)
# The first(end of file) block will be short, since this leaves
# the rest aligned on a blocksize boundary. This may be more
# efficient than having the last (first in file) block be short
f.seek(-lastblock,2)
yield f.read(lastblock)
for i in range(fullblocks-1,-1, -1):
f.seek(i * blocksize)
yield f.read(blocksize)
def tail(f, nlines):
buf = ''
result = []
for block in rblocks(f):
buf = block + buf
lines = buf.splitlines()
# Return all lines except the first (since may be partial)
if lines:
result.extend(lines[1:]) # First line may not be complete
if(len(result) >= nlines):
return result[-nlines:]
buf = lines[0]
return ([buf]+result)[-nlines:]
f=open('file_to_tail.txt','rb')
for line in tail(f, 20):
print line
편집] 더 구체적인 버전이 추가되었습니다 (두 번 반전해야합니다)
F.seek (0, 2)로 파일 끝으로 이동 한 다음 readline ()을 대체하는 다음과 같은 줄을 하나씩 읽을 수 있습니다.
def readline_backwards(self, f):
backline = ''
last = ''
while not last == '\n':
backline = last + backline
if f.tell() <= 0:
return backline
f.seek(-1, 1)
last = f.read(1)
f.seek(-1, 1)
backline = last
last = ''
while not last == '\n':
backline = last + backline
if f.tell() <= 0:
return backline
f.seek(-1, 1)
last = f.read(1)
f.seek(-1, 1)
f.seek(1, 1)
return backline
Eyecue Answer (Jun 10 '10 at 21:28)를 기반으로 :이 클래스는 객체를 파일에 추가하여 Head () 및 Tail () 메소드를 추가합니다.
class File(file):
def head(self, lines_2find=1):
self.seek(0) #Rewind file
return [self.next() for x in xrange(lines_2find)]
def tail(self, lines_2find=1):
self.seek(0, 2) #go to end of file
bytes_in_file = self.tell()
lines_found, total_bytes_scanned = 0, 0
while (lines_2find+1 > lines_found and
bytes_in_file > total_bytes_scanned):
byte_block = min(1024, bytes_in_file-total_bytes_scanned)
self.seek(-(byte_block+total_bytes_scanned), 2)
total_bytes_scanned += byte_block
lines_found += self.read(1024).count('\n')
self.seek(-total_bytes_scanned, 2)
line_list = list(self.readlines())
return line_list[-lines_2find:]
용법:
f = File('path/to/file', 'r')
f.head(3)
f.tail(3)
이러한 솔루션 중 일부에는 파일이 n으로 끝나지 않거나 전체 첫 번째 줄을 읽을 수있는 경우 문제가 있습니다.
def tail(file, n=1, bs=1024):
f = open(file)
f.seek(-1,2)
l = 1-f.read(1).count('\n') # If file doesn't end in \n, count it anyway.
B = f.tell()
while n >= l and B > 0:
block = min(bs, B)
B -= block
f.seek(B, 0)
l += f.read(block).count('\n')
f.seek(B, 0)
l = min(l,n) # discard first (incomplete) line if l > n
lines = f.readlines()[-l:]
f.close()
return lines
파일의 마지막 줄에서 특정 값을 읽고이 스레드를 우연히 발견해야했습니다. 파이썬에서 휠을 재창조하는 대신/usr/local/bin/get_last_netp로 저장된 작은 쉘 스크립트로 끝났습니다.
#! /bin/bash
tail -n1 /home/leif/projects/transfer/export.log | awk {'print $14'}
그리고 파이썬 프로그램에서 :
from subprocess import check_output
last_netp = int(check_output("/usr/local/bin/get_last_netp"))
Deque를 사용하는 첫 번째 예제는 아니지만 단순한 예제입니다. 이것은 일반적입니다. 파일뿐만 아니라 반복 가능한 객체에서 작동합니다.
#!/usr/bin/env python
import sys
import collections
def tail(iterable, N):
deq = collections.deque()
for thing in iterable:
if len(deq) >= N:
deq.popleft()
deq.append(thing)
for thing in deq:
yield thing
if __name__ == '__main__':
for line in tail(sys.stdin,10):
sys.stdout.write(line)
This is my version of tailf
import sys, time, os
filename = 'path to file'
try:
with open(filename) as f:
size = os.path.getsize(filename)
if size < 1024:
s = size
else:
s = 999
f.seek(-s, 2)
l = f.read()
print l
while True:
line = f.readline()
if not line:
time.sleep(1)
continue
print line
except IOError:
pass
import time
attemps = 600
wait_sec = 5
fname = "YOUR_PATH"
with open(fname, "r") as f:
where = f.tell()
for i in range(attemps):
line = f.readline()
if not line:
time.sleep(wait_sec)
f.seek(where)
else:
print line, # already has newline
import itertools
fname = 'log.txt'
offset = 5
n = 10
with open(fname) as f:
n_last_lines = list(reversed([x for x in itertools.islice(f, None)][-(offset+1):-(offset+n+1):-1]))
abc = "2018-06-16 04:45:18.68"
filename = "abc.txt"
with open(filename) as myFile:
for num, line in enumerate(myFile, 1):
if abc in line:
lastline = num
print "last occurance of work at file is in "+str(lastline)
There is very useful module that can do this:
from file_read_backwards import FileReadBackwards
with FileReadBackwards("/tmp/file", encoding="utf-8") as frb:
# getting lines by lines starting from the last line up
for l in frb:
print(l)
On second thought, this is probably just as fast as anything here.
def tail( f, window=20 ):
lines= ['']*window
count= 0
for l in f:
lines[count%window]= l
count += 1
print lines[count%window:], lines[:count%window]
It's a lot simpler. And it does seem to rip along at a good pace.
I found a probably the easiest way to find the first or last N lines of a file
Last N lines of a file(For Ex:N=10)
file=open("xyz.txt",'r")
liner=file.readlines()
for ran in range((len(liner)-N),len(liner)):
print liner[ran]
First N lines of a file(For Ex:N=10)
file=open("xyz.txt",'r")
liner=file.readlines()
for ran in range(0,N+1):
print liner[ran]
it's so simple:
def tail(fname,nl):
with open(fname) as f:
data=f.readlines() #readlines return a list
print(''.join(data[-nl:]))
Although this isn't really on the efficient side with big files, this code is pretty straight-forward:
- It reads the file object,
f. - It splits the string returned using newlines,
\n. It gets the array lists last indexes, using the negative sign to stand for the last indexes, and the
:to get a subarray.def tail(f,n): return "\n".join(f.read().split("\n")[-n:])
제휴하지 않습니다 StackOverflow
