Я впервые сталкиваюсь с этой ошибкой и не могу разобраться((
import multiprocessing as mp
from rich.console import Console
from rich.prompt import Confirm
from rich.table import Table
from typing import List
console = Console()
class Unique:
def __init__(self) -> None:
self.unique_lines_count: int = 0
self.not_unique_lines_count: int = 0
self.data_count: int = 0
self.lines_count: int = 0
self.unique_lines: List[str] = []
self.is_write_unique_lines: bool
self.pool = mp.Pool(processes=8)
self.jobs = []
self.start()
def check_unique(self) -> None:
with open('data/new_data.txt', encoding='utf-8') as new_data:
next_line_byte = new_data.tell()
for line in new_data:
self.jobs.append(self.pool.apply_async(self.check_line, next_line_byte))
next_line_byte = new_data.tell()
# self.check_line(line_id)
def check_line(self, new_line_byte) -> None:
with open('data/new_data.txt', encoding='utf-8') as new_data:
console.log('Проверяю строку')
new_data.seek(new_line_byte)
line = new_data.readline()
line_id_db = False
with open('data/old_data.txt', encoding='utf-8') as old_data:
for old_line in old_data:
if line == old_line:
line_id_db = True
if line_id_db is False:
self.unique_lines.append(line)
console.log('Проверил строку')
def write_unique_lines(self) -> None:
with open('data/unique_data.txt', 'w', encoding='utf-8') as file:
for line in self.unique_lines:
file.write(line)
file.close()
def show_stats(self):
table = Table(title="Статистика")
table.add_column("Тип", style="cyan", no_wrap=True)
table.add_column("Количество", style="magenta")
table.add_row("Уникальных строк", str(len(self.unique_lines)))
console.print('\n')
console.print(table)
def start(self) -> None:
console.log('Программа запущена')
is_write_unique_lines = Confirm.ask("Добавлять уникальные в базу?")
console.log('Начинаю сортировку')
self.check_unique()
for job in self.jobs:
job.get()
self.pool.close()
if is_write_unique_lines is True:
self.write_unique_lines()
console.log('Записал уникальные данные')
self.show_stats()
input('Нажмите ентер чтобы закрыть')
if __name__ == '__main__':
Unique()
Traceback (most recent call last):
File "C:\Users\Fox\Desktop\Новая папка (4)\main.py", line 89, in
Unique()
File "C:\Users\Fox\Desktop\Новая папка (4)\main.py", line 26, in init
self.start()
File "C:\Users\Fox\Desktop\Новая папка (4)\main.py", line 73, in start
self.check_unique()
File "C:\Users\Fox\Desktop\Новая папка (4)\main.py", line 33, in check_unique
next_line_byte = new_data.tell()
OSError: telling position disabled by next() call
